282 lines
11 KiB
Python
282 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPTS = Path(__file__).resolve().parent
|
|
BOOTSTRAP = SCRIPTS / "bootstrap-legacy-public-runtime-upload-root.ps1"
|
|
|
|
|
|
def _function(source: str, name: str, next_name: str) -> str:
|
|
start = source.index(f"function {name}")
|
|
end = source.index(f"function {next_name}", start)
|
|
return source[start:end]
|
|
|
|
|
|
class PublicRuntimeEnvironmentHandoffTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.source = BOOTSTRAP.read_text(encoding="utf-8")
|
|
cls.functions = "\n".join(
|
|
(
|
|
_function(
|
|
cls.source,
|
|
"Save-CompleteProcessEnvironment",
|
|
"Get-IdentityEnvironmentValue",
|
|
),
|
|
_function(
|
|
cls.source,
|
|
"Invoke-EffectiveApiSettingsProbe",
|
|
"Set-RequiredApiEnvironmentFromIdentity",
|
|
),
|
|
_function(
|
|
cls.source,
|
|
"Get-FutureLauncherRequiredApiSettings",
|
|
"Assert-RequiredApiSettingsEqual",
|
|
),
|
|
_function(
|
|
cls.source,
|
|
"Assert-RequiredApiSettingsEqual",
|
|
"Assert-EnvironmentFilePinned",
|
|
),
|
|
_function(
|
|
cls.source,
|
|
"Set-CompleteProcessEnvironment",
|
|
"ConvertTo-WindowsCommandLineArgument",
|
|
),
|
|
)
|
|
)
|
|
|
|
def _powershell(self) -> str:
|
|
powershell = shutil.which("powershell.exe")
|
|
if powershell is None:
|
|
self.skipTest("Windows PowerShell 5.1 is not available")
|
|
return powershell
|
|
|
|
@staticmethod
|
|
def _write_fake_api(api_root: Path, *, client_secret: str) -> None:
|
|
package = api_root / "app"
|
|
package.mkdir(parents=True)
|
|
(package / "__init__.py").write_text("", encoding="utf-8")
|
|
(package / "config.py").write_text(
|
|
"""import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
|
|
values = {}
|
|
for line in Path('.env').read_text(encoding='utf-8').splitlines():
|
|
if line and not line.lstrip().startswith('#') and '=' in line:
|
|
key, value = line.split('=', 1)
|
|
values[key.strip()] = value.strip()
|
|
|
|
|
|
def get(name, default=''):
|
|
return os.environ.get(name, values.get(name, default))
|
|
|
|
|
|
class Secret:
|
|
def __init__(self, value): self.value = value
|
|
def get_secret_value(self): return self.value
|
|
|
|
|
|
class Settings:
|
|
database_url = get('DATABASE_URL')
|
|
session_secret = get('SESSION_SECRET')
|
|
engine_gateway_shared_secret = Secret(get('ENGINE_GATEWAY_SHARED_SECRET'))
|
|
engine_url = get('ENGINE_URL')
|
|
oauth_google_client_id = get('OAUTH_GOOGLE_CLIENT_ID')
|
|
oauth_google_client_secret = get('OAUTH_GOOGLE_CLIENT_SECRET')
|
|
oauth_redirect_uri = get('OAUTH_REDIRECT_URI')
|
|
auth_allowed_email_domains = json.loads(get('AUTH_ALLOWED_EMAIL_DOMAINS', '[]'))
|
|
auth_teacher_emails = json.loads(get('AUTH_TEACHER_EMAILS', '[]'))
|
|
auth_admin_emails = json.loads(get('AUTH_ADMIN_EMAILS', '[]'))
|
|
auth_super_admin_emails = json.loads(get('AUTH_SUPER_ADMIN_EMAILS', '[]'))
|
|
auth_approved_emails = json.loads(get('AUTH_APPROVED_EMAILS', '[]'))
|
|
auth_new_user_default_status = get('AUTH_NEW_USER_DEFAULT_STATUS', 'pending')
|
|
auth_email_cohort_map = json.loads(get('AUTH_EMAIL_COHORT_MAP', '{}'))
|
|
auth_domain_cohort_map = json.loads(get('AUTH_DOMAIN_COHORT_MAP', '{}'))
|
|
default_affiliation = get('DEFAULT_AFFILIATION')
|
|
|
|
|
|
settings = Settings()
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
(api_root / ".env").write_text(
|
|
"\n".join(
|
|
(
|
|
"DATABASE_URL=postgresql://unit:pw@127.0.0.1:5432/vignette",
|
|
"SESSION_SECRET=unit-session-secret-that-is-long-enough",
|
|
"ENGINE_GATEWAY_SHARED_SECRET=unit-engine-secret-that-is-long-enough",
|
|
"ENGINE_URL=http://127.0.0.1:9099",
|
|
"OAUTH_GOOGLE_CLIENT_ID=unit-client.apps.googleusercontent.com",
|
|
f"OAUTH_GOOGLE_CLIENT_SECRET={client_secret}",
|
|
"OAUTH_REDIRECT_URI=https://api-vignette.chanpaca.net/auth/callback",
|
|
'AUTH_ALLOWED_EMAIL_DOMAINS=["example.test"]',
|
|
'AUTH_SUPER_ADMIN_EMAILS=["owner@example.test"]',
|
|
"AUTH_NEW_USER_DEFAULT_STATUS=approved",
|
|
"DEFAULT_AFFILIATION=unit",
|
|
"",
|
|
)
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def _run_handoff(self, *, mismatched_target: bool) -> subprocess.CompletedProcess[str]:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-env-handoff-") as raw:
|
|
root = Path(raw)
|
|
prior_api = root / "prior" / "apps" / "api"
|
|
stable_root = root / "stable"
|
|
target_api = stable_root / "apps" / "api"
|
|
self._write_fake_api(prior_api, client_secret="unit-prior-secret")
|
|
self._write_fake_api(
|
|
target_api,
|
|
client_secret=(
|
|
"unit-target-drift" if mismatched_target else "unit-prior-secret"
|
|
),
|
|
)
|
|
harness = root / "handoff.ps1"
|
|
escaped_prior = str(prior_api).replace("'", "''")
|
|
escaped_stable = str(stable_root).replace("'", "''")
|
|
escaped_python = sys.executable.replace("'", "''")
|
|
harness.write_text(
|
|
self.functions
|
|
+ f"""
|
|
$resolvedPythonPath = '{escaped_python}'
|
|
$resolvedStableSourceRoot = '{escaped_stable}'
|
|
$required = @(
|
|
'DATABASE_URL','SESSION_SECRET','ENGINE_URL','ENGINE_GATEWAY_SHARED_SECRET',
|
|
'OAUTH_GOOGLE_CLIENT_ID','OAUTH_GOOGLE_CLIENT_SECRET','OAUTH_REDIRECT_URI',
|
|
'AUTH_ALLOWED_EMAIL_DOMAINS','AUTH_TEACHER_EMAILS','AUTH_ADMIN_EMAILS',
|
|
'AUTH_SUPER_ADMIN_EMAILS','AUTH_APPROVED_EMAILS','AUTH_NEW_USER_DEFAULT_STATUS',
|
|
'AUTH_EMAIL_COHORT_MAP','AUTH_DOMAIN_COHORT_MAP','DEFAULT_AFFILIATION'
|
|
)
|
|
$completeEnvironment = Save-CompleteProcessEnvironment
|
|
$priorEnvironment = [ordered]@{{}}
|
|
foreach ($key in $completeEnvironment.Keys) {{
|
|
$isRequired = $false
|
|
foreach ($requiredName in $required) {{
|
|
if ([string]$key -ieq $requiredName) {{
|
|
$isRequired = $true
|
|
break
|
|
}}
|
|
}}
|
|
if (-not $isRequired) {{
|
|
$priorEnvironment[[string]$key] = [string]$completeEnvironment[$key]
|
|
}}
|
|
}}
|
|
$priorIdentity = [ordered]@{{
|
|
cwd = '{escaped_prior}'
|
|
executable_path = '{escaped_python}'
|
|
environment = $priorEnvironment
|
|
}}
|
|
foreach ($requiredName in $required) {{
|
|
if ($priorIdentity.environment.Contains($requiredName)) {{
|
|
throw 'Test fixture unexpectedly contains required PID environment'
|
|
}}
|
|
}}
|
|
$prior = Invoke-EffectiveApiSettingsProbe `
|
|
-PythonPath $resolvedPythonPath `
|
|
-ApiCwd $priorIdentity.cwd `
|
|
-Environment $priorIdentity.environment
|
|
$future = Get-FutureLauncherRequiredApiSettings `
|
|
-PriorApiIdentity $priorIdentity `
|
|
-RequiredNames $required `
|
|
-EnginePort 9099
|
|
Assert-RequiredApiSettingsEqual `
|
|
-Expected $prior `
|
|
-Actual $future `
|
|
-Role 'future launcher'
|
|
Write-Output 'PASS'
|
|
""",
|
|
encoding="utf-8-sig",
|
|
)
|
|
return subprocess.run(
|
|
[
|
|
self._powershell(),
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(harness),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
|
|
def test_pid_environment_empty_dotenv_only_reboots_to_same_effective_settings(
|
|
self,
|
|
) -> None:
|
|
completed = self._run_handoff(mismatched_target=False)
|
|
self.assertEqual(
|
|
0,
|
|
completed.returncode,
|
|
msg=f"stdout={completed.stdout}\nstderr={completed.stderr}",
|
|
)
|
|
self.assertEqual("PASS", completed.stdout.strip())
|
|
self.assertNotIn("unit-prior-secret", completed.stdout + completed.stderr)
|
|
|
|
def test_future_dotenv_drift_fails_without_printing_secret(self) -> None:
|
|
completed = self._run_handoff(mismatched_target=True)
|
|
self.assertNotEqual(0, completed.returncode)
|
|
self.assertNotIn("unit-prior-secret", completed.stdout + completed.stderr)
|
|
self.assertNotIn("unit-target-drift", completed.stdout + completed.stderr)
|
|
self.assertIn("effective settings digest drift", completed.stderr)
|
|
|
|
def test_source_hash_is_pinned_before_settings_copy_and_after_copy(self) -> None:
|
|
main = self.source[
|
|
self.source.index("$priorEnvironmentFilePath =") :
|
|
self.source.index("$priorTunnelIdentity =", self.source.index("$priorEnvironmentFilePath ="))
|
|
]
|
|
source_hash = main.index("$priorEnvironmentFileSha256 =")
|
|
settings = main.index("Set-RequiredApiEnvironmentFromIdentity")
|
|
copy = main.index("Ensure-ReleaseEnvironmentFile")
|
|
future = main.index("Get-FutureLauncherRequiredApiSettings")
|
|
equality = main.index("Assert-RequiredApiSettingsEqual")
|
|
self.assertLess(source_hash, settings)
|
|
self.assertLess(settings, copy)
|
|
self.assertLess(copy, future)
|
|
self.assertLess(future, equality)
|
|
ensure = self.source[
|
|
self.source.index("function Ensure-ReleaseEnvironmentFile") :
|
|
self.source.index("function Get-FutureLauncherRequiredApiSettings")
|
|
]
|
|
self.assertIn('check-ignore --quiet -- "apps/api/.env"', ensure)
|
|
self.assertGreaterEqual(ensure.count("Assert-EnvironmentFilePinned `"), 3)
|
|
|
|
def test_future_launcher_uses_the_actual_engine_port(self) -> None:
|
|
function = self.source[
|
|
self.source.index("function Get-FutureLauncherRequiredApiSettings") :
|
|
self.source.index("function Assert-RequiredApiSettingsEqual")
|
|
]
|
|
self.assertIn("[int]$EnginePort", function)
|
|
self.assertIn(
|
|
'$futureEnvironment["ENGINE_URL"] = "http://127.0.0.1:$EnginePort"',
|
|
function,
|
|
)
|
|
self.assertNotIn("127.0.0.1:3001", function)
|
|
main = self.source[
|
|
self.source.index("$priorEnvironmentFilePath =") :
|
|
self.source.index(
|
|
"$priorTunnelIdentity =",
|
|
self.source.index("$priorEnvironmentFilePath ="),
|
|
)
|
|
]
|
|
self.assertIn("-EnginePort $EnginePort", main)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|