import json import os import stat import subprocess import sys import tempfile import unittest from pathlib import Path PROJECT = Path(__file__).resolve().parents[1] class ProjectScaffoldTests(unittest.TestCase): def test_colab_notebook_is_valid_plain_python(self): result = subprocess.run( [ sys.executable, "tools/run_colab_notebook.py", "notebooks/quant_os_colab.ipynb", ], cwd=PROJECT, check=True, capture_output=True, text=True, ) payload = json.loads(result.stdout.strip().splitlines()[-1]) self.assertTrue(payload["ok"]) self.assertGreaterEqual(payload["code_cells"], 4) self.assertFalse(payload["executed"]) optional_result = subprocess.run( [ sys.executable, "tools/run_colab_notebook.py", "notebooks/quant_os_colab.ipynb", "--include-optional", ], cwd=PROJECT, check=True, capture_output=True, text=True, ) optional_payload = json.loads( optional_result.stdout.strip().splitlines()[-1] ) self.assertTrue(optional_payload["optional_included"]) self.assertGreater( optional_payload["code_cells"], payload["code_cells"], ) def test_secret_files_are_ignored(self): ignore = (PROJECT / ".gitignore").read_text(encoding="utf-8") for entry in ( ".env", "secrets/", "credentials/", "userdata_mini/", "data/", "build/", ): self.assertIn(entry, ignore) def test_tracked_secret_scan_passes(self): result = subprocess.run( [sys.executable, "tools/check_tracked_secrets.py"], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("tracked secret scan: OK", result.stdout) def test_claude_project_contract_exists(self): content = (PROJECT / "CLAUDE.md").read_text(encoding="utf-8") self.assertIn("executable source of truth for Quant OS", content) self.assertIn("make local", content) def test_gitea_workflow_is_at_independent_repo_root(self): workflow = PROJECT / ".gitea/workflows/ci.yml" self.assertTrue(workflow.is_file()) content = workflow.read_text(encoding="utf-8") self.assertIn("bash scripts/validate_all.sh", content) self.assertNotIn("working-directory: quant-os", content) self.assertNotIn('"quant-os/**"', content) self.assertIn(r"\.env(\.[^/]+)?", content) self.assertIn(r"\.env\.example", content) def test_long_lived_product_namespace_is_installable(self): pyproject = (PROJECT / "pyproject.toml").read_text(encoding="utf-8") self.assertIn('name = "quant-os"', pyproject) self.assertIn('quant-os = "quant_os.cli:main"', pyproject) self.assertIn('quant60 = "quant60.cli:main"', pyproject) self.assertTrue((PROJECT / "src/quant_os/__init__.py").is_file()) def test_cli_can_resolve_an_explicit_checkout_from_elsewhere(self): environment = {**os.environ, "PYTHONPATH": str(PROJECT / "src")} with tempfile.TemporaryDirectory() as temporary: result = subprocess.run( [ sys.executable, "-m", "quant_os", "--project-root", str(PROJECT), "doctor", ], cwd=temporary, env=environment, check=False, capture_output=True, text=True, ) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) self.assertTrue(payload["ok"]) self.assertEqual( payload["architecture"]["scope"], "new_namespace_only", ) self.assertGreater( payload["migration"]["legacy_python_files_outside_checker"], 0, ) def test_nginx_source_of_truth_has_atomic_release_routes(self): config = ( PROJECT / "ops/nginx/quant-os-static.conf" ).read_text(encoding="utf-8") self.assertIn("location = /quant-os/status/ {", config) self.assertIn( "rewrite ^ /quant-os/status/index.html break;", config, ) self.assertIn("location ^~ /quant-os/status/", config) self.assertIn("root /srv/www;", config) self.assertNotIn("index index.html;", config) self.assertNotIn("try_files $uri $uri/", config) self.assertIn( "https://$host/quant-os/status/", config, ) self.assertNotIn("/quants-strategies/", config) self.assertNotIn(":8443", config) temporary = ( PROJECT / "ops/nginx/quant-os-legacy-redirects-302.conf" ).read_text(encoding="utf-8") stable = ( PROJECT / "ops/nginx/quant-os-legacy-redirects-308.conf" ).read_text(encoding="utf-8") self.assertIn( "location = /quants-strategies/quant-os/status", temporary, ) self.assertIn("return 302 ", temporary) self.assertNotIn("return 308 ", temporary) self.assertIn("return 308 ", stable) def test_static_status_deployer_is_atomic_and_non_overwriting(self): sha = "a" * 40 with tempfile.TemporaryDirectory() as temporary: web_root = Path(temporary) / "web" result = subprocess.run( [ "bash", "ops/deploy_static_status.sh", sha, "status", str(web_root), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual( (web_root / "status").readlink(), Path("releases") / sha, ) release = web_root / "releases" / sha self.assertEqual( {path.name for path in release.iterdir()}, {"index.html", "status-data.js"}, ) self.assertFalse(release.stat().st_mode & stat.S_IWUSR) self.assertFalse( release.joinpath("index.html").stat().st_mode & stat.S_IWUSR ) self.assertFalse(web_root.joinpath(f".incoming-{sha}").exists()) self.assertFalse(web_root.joinpath(".status-deploy.lock").exists()) second_sha = "b" * 40 switched = subprocess.run( [ "bash", "ops/deploy_static_status.sh", second_sha, "status", str(web_root), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertEqual(switched.returncode, 0, switched.stderr) self.assertEqual( (web_root / "status").readlink(), Path("releases") / second_sha, ) second = subprocess.run( [ "bash", "ops/deploy_static_status.sh", second_sha, "status", str(web_root), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertNotEqual(second.returncode, 0) self.assertIn("refusing to overwrite", second.stderr) def test_static_status_deployer_rejects_non_regular_or_extra_source(self): cases = ("symlink", "extra-directory") for case in cases: with self.subTest(case=case): temporary_context = tempfile.TemporaryDirectory() self.addCleanup(temporary_context.cleanup) temporary = temporary_context.name root = Path(temporary) source = root / "source" source.mkdir() source.joinpath("status-data.js").write_text( "window.STATUS = {};\n", encoding="utf-8", ) if case == "symlink": source.joinpath("index.html").symlink_to( PROJECT / "status/index.html" ) else: source.joinpath("index.html").write_text( "\n", encoding="utf-8", ) source.joinpath("extra").mkdir() web_root = root / "web" result = subprocess.run( [ "bash", "ops/deploy_static_status.sh", "c" * 40, str(source), str(web_root), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertNotEqual(result.returncode, 0) self.assertFalse(web_root.joinpath("releases").exists()) def test_static_status_deploy_and_activation_share_concurrency_lock(self): sha = "d" * 40 with tempfile.TemporaryDirectory() as temporary: web_root = Path(temporary) / "web" web_root.mkdir() web_root.joinpath(".status-deploy.lock").mkdir() deploy = subprocess.run( [ "bash", "ops/deploy_static_status.sh", sha, "status", str(web_root), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertNotEqual(deploy.returncode, 0) self.assertIn("another status deployment", deploy.stderr) release = web_root / "releases" / sha release.mkdir(parents=True) for name in ("index.html", "status-data.js"): release.joinpath(name).write_text(name, encoding="utf-8") activate = subprocess.run( [ "bash", "ops/activate_static_status.sh", sha, str(web_root), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertNotEqual(activate.returncode, 0) self.assertIn("another status deployment", activate.stderr) self.assertFalse(web_root.joinpath("status").exists()) def test_static_status_activation_rejects_extra_release_entry(self): sha = "e" * 40 with tempfile.TemporaryDirectory() as temporary: web_root = Path(temporary) / "web" release = web_root / "releases" / sha release.mkdir(parents=True) for name in ("index.html", "status-data.js"): release.joinpath(name).write_text(name, encoding="utf-8") release.joinpath("extra").mkdir() result = subprocess.run( [ "bash", "ops/activate_static_status.sh", sha, str(web_root), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertNotEqual(result.returncode, 0) self.assertIn("exactly index.html and status-data.js", result.stderr) self.assertFalse(web_root.joinpath("status").exists()) self.assertFalse(web_root.joinpath(".status-deploy.lock").exists()) def test_local_asset_inventory_is_deterministic_and_path_safe(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / "data").mkdir() (root / "artifacts").mkdir() (root / "data" / "one.bin").write_bytes(b"one") (root / "artifacts" / "two.json").write_text( "{}\n", encoding="utf-8", ) first = root / "first.json" result = subprocess.run( [ sys.executable, str(PROJECT / "tools/inventory_local_assets.py"), str(root), "--output", str(first), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(first.read_text(encoding="utf-8")) self.assertEqual( [entry["path"] for entry in payload["entries"]], ["data/one.bin", "artifacts/two.json"], ) encoded = json.dumps(payload) self.assertNotIn(str(root), encoded) compare = subprocess.run( [ sys.executable, str(PROJECT / "tools/inventory_local_assets.py"), str(root), "--compare", str(first), ], cwd=PROJECT, check=False, capture_output=True, text=True, ) self.assertEqual(compare.returncode, 0, compare.stderr) if __name__ == "__main__": unittest.main()