douyin-sparkflow/DouYinSparkFlow/tests/test_deployment_contract.py

import unittest
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[2]
SOURCE_ROOT = REPO_ROOT / "DouYinSparkFlow"


class DeploymentContractTests(unittest.TestCase):
    def test_github_workflow_is_at_repository_root(self):
        workflow = REPO_ROOT / ".github" / "workflows" / "schedule.yml"
        self.assertTrue(workflow.is_file())
        self.assertFalse((SOURCE_ROOT / ".github" / "workflows" / "schedule.yml").exists())
        text = workflow.read_text(encoding="utf-8")
        self.assertIn("working-directory: DouYinSparkFlow", text)
        self.assertIn("SPARKFLOW_BROWSER_PROFILE_ROOT", text)
        self.assertIn("SPARKFLOW_MANUAL_RUN", text)
        self.assertIn('USER_DATA: "[]"', text)
        self.assertIn("run_task:", text)
        self.assertGreaterEqual(text.count("github.event_name == 'schedule' || inputs.run_task"), 2)
        self.assertIn("path: DouYinSparkFlow/logs/", text)

    def test_github_actions_are_pinned_to_commit_shas(self):
        import re

        workflow = (REPO_ROOT / ".github" / "workflows" / "schedule.yml").read_text(encoding="utf-8")
        uses_values = re.findall(r"^\s*-?\s*uses:\s*([^#\s]+)", workflow, flags=re.MULTILINE)
        self.assertTrue(uses_values)
        for value in uses_values:
            self.assertRegex(value, r"^[^@]+@[0-9a-f]{40}$")

    def test_runtime_config_is_not_tracked_as_the_template(self):
        self.assertTrue((SOURCE_ROOT / "config.example.json").is_file())
        self.assertIn("config.json", (SOURCE_ROOT / ".gitignore").read_text(encoding="utf-8"))

    def test_scheduler_uses_local_task_runner_and_migrates_legacy_commands(self):
        compose = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8")
        runner_path = SOURCE_ROOT / "scripts" / "run_scheduled_task.sh"
        runner = runner_path.read_text(encoding="utf-8")
        cron_runner = (SOURCE_ROOT / "scripts" / "cron_runner.py").read_text(encoding="utf-8")
        scheduler = compose.split("  scheduler:", 1)[1].split("\n  task:", 1)[0]

        self.assertTrue(runner_path.is_file())
        self.assertIn("python main.py --doTask", runner)
        self.assertIn("migrate_legacy_crontab", cron_runner)
        self.assertNotIn("/var/run/docker.sock:/var/run/docker.sock", scheduler)

    def test_compose_runtime_mounts_follow_least_privilege(self):
        text = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8")
        web = text.split("  web:", 1)[1].split("\n  login-desktop:", 1)[0]
        scheduler = text.split("  scheduler:", 1)[1].split("\n  task:", 1)[0]
        task = text.split("  task:", 1)[1]

        self.assertIn("/var/run/docker.sock:/var/run/docker.sock", web)
        self.assertIn(".:/opt/douyin-sparkflow", web)
        self.assertNotIn("/var/run/docker.sock:/var/run/docker.sock", scheduler)
        self.assertNotIn(".:/opt/douyin-sparkflow", scheduler)
        self.assertNotIn("/var/run/docker.sock:/var/run/docker.sock", task)
        self.assertNotIn(".:/opt/douyin-sparkflow", task)
        for service in (scheduler, task):
            self.assertIn(
                "./state/browser-profiles:/opt/douyin-sparkflow/state/browser-profiles",
                service,
            )

    def test_cron_reader_accepts_windows_utf8_bom(self):
        import tempfile

        from scripts.cron_runner import read_crontab

        with tempfile.TemporaryDirectory() as temp_dir:
            path = Path(temp_dir) / "root"
            path.write_text(
                "*/20 10-17 * * * cd /app && python main.py --doTask\n",
                encoding="utf-8-sig",
            )
            lines = read_crontab(path)

        self.assertEqual(len(lines), 1)
        self.assertTrue(lines[0].startswith("*/20 "))

    def test_legacy_cron_command_migrates_without_touching_schedule(self):
        from scripts.cron_runner import migrate_legacy_line

        legacy = (
            "20 18 * * * /bin/bash -lc 'docker ps --format \"{{.Names}}\" | "
            "grep douyin-web; docker exec douyin-web sh -lc \"cd /app && "
            "env SPARKFLOW_MANUAL_RUN=1 SPARKFLOW_MANUAL_UNSENT_ONLY=1 "
            "python main.py --doTask\"' >> /var/log/douyin-sparkflow.log 2>&1"
        )
        migrated = migrate_legacy_line(legacy)

        self.assertTrue(migrated.startswith("20 18 * * * "))
        self.assertIn("run_scheduled_task.sh", migrated)
        self.assertIn("SPARKFLOW_MANUAL_UNSENT_ONLY=1", migrated)
        self.assertNotIn("docker ps", migrated)
        self.assertNotIn("docker exec", migrated)

    def test_build_proxy_does_not_leak_into_runtime_and_runtime_proxy_is_explicit(self):
        dockerfile = (SOURCE_ROOT / "Dockerfile.server").read_text(encoding="utf-8")
        compose = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8")
        self.assertIn("# Build proxies must never leak", dockerfile)
        self.assertIn("http_proxy=", dockerfile)
        self.assertIn("HTTP_PROXY: ${HTTP_PROXY_BUILD:-}", compose)
        self.assertIn("HTTPS_PROXY: ${HTTPS_PROXY_BUILD:-}", compose)
        self.assertIn("ALL_PROXY: ${ALL_PROXY_BUILD:-}", compose)
        for service in ("web", "scheduler", "task"):
            block = compose.split(f"  {service}:", 1)[1]
            block = block.split("\n  ", 1)[0]
            self.assertNotIn("HTTP_PROXY: http://proxy:7890", block)
            self.assertNotIn("http_proxy: http://proxy:7890", block)

    def test_sensitive_ports_bind_to_loopback_by_default(self):
        text = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8")
        self.assertIn("${PROXY_BIND_ADDRESS:-127.0.0.1}:${PROXY_HTTP_PORT:-7890}:7890", text)
        self.assertIn(
            "${LOGIN_DESKTOP_BIND_ADDRESS:-127.0.0.1}:${LOGIN_DESKTOP_WEB_PORT:-8788}:6080",
            text,
        )

    def test_container_login_api_and_public_url_are_wired(self):
        text = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8")
        self.assertIn("SPARKFLOW_LOGIN_DESKTOP_API_URL: http://login-desktop:18090", text)
        self.assertIn("SPARKFLOW_LOGIN_DESKTOP_PUBLIC_URL", text)
        self.assertIn("/login-desktop/proxy/vnc.html", text)
        self.assertIn("SPARKFLOW_LOGIN_DESKTOP_NOVNC_WS_URL", text)

    def test_installers_preserve_runtime_config_and_do_not_require_bash_on_windows(self):
        server = (REPO_ROOT / "deploy" / "install-server.sh").read_text(encoding="utf-8")
        windows = (REPO_ROOT / "deploy" / "install-local.ps1").read_text(encoding="utf-8")
        self.assertIn("runtime_config_backup", server)
        self.assertIn("Restored runtime config.json", server)
        self.assertIn("remove_legacy_host_cron", server)
        self.assertIn("host-crontab-", server)
        self.assertIn("docker ps --format", server)
        self.assertNotIn("bash ./refresh_proxy.sh", windows)
        self.assertIn("Initialize-ProxyConfig", windows)

    def test_playwright_base_image_argument_is_used(self):
        dockerfile = (SOURCE_ROOT / "Dockerfile.server").read_text(encoding="utf-8")
        self.assertTrue(dockerfile.startswith("ARG PLAYWRIGHT_BASE_IMAGE="))
        self.assertIn("FROM ${NODE_RUNTIME_IMAGE} AS node-runtime", dockerfile)
        self.assertIn("FROM ${PLAYWRIGHT_BASE_IMAGE}", dockerfile)
        self.assertIn("COPY --from=node-runtime /usr/local/bin/node", dockerfile)
        self.assertIn("docker.io", dockerfile)
        self.assertIn("node --version", dockerfile)
        self.assertNotIn("github.com/docker/compose", dockerfile)

    def test_docker_build_context_excludes_runtime_data(self):
        dockerignore = (SOURCE_ROOT / ".dockerignore").read_text(encoding="utf-8")
        for entry in ("logs/", "config.json", "usersData.json", "webui_settings.json"):
            self.assertIn(entry, dockerignore)

    def test_login_desktop_uses_direct_first_network_route(self):
        compose = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8")
        env_example = (REPO_ROOT / ".env.example").read_text(encoding="utf-8")
        server = (SOURCE_ROOT / "login_desktop_server.py").read_text(encoding="utf-8")
        login_block = compose.split("  login-desktop:", 1)[1].split("  scheduler:", 1)[0]
        self.assertIn("LOGIN_DESKTOP_PROXY_MODE: ${LOGIN_DESKTOP_PROXY_MODE:-auto}", login_block)
        self.assertIn("LOGIN_DESKTOP_PROXY: ${LOGIN_DESKTOP_PROXY:-http://proxy:7890}", login_block)
        self.assertNotIn("HTTP_PROXY: http://proxy:7890", login_block)
        self.assertIn("LOGIN_DESKTOP_PROXY_MODE=auto", env_example)
        self.assertIn('candidates.append(("direct", None))', server)
        self.assertIn('candidates.append(("proxy", LOGIN_PROXY_SERVER))', server)
        self.assertIn('"--no-proxy-server"', server)
        self.assertIn('"/preflight"', server)
        self.assertIn('LOGIN_DESKTOP_PROXY_MODE: ${LOGIN_DESKTOP_PROXY_MODE:-auto}', login_block)
        dashboard = (SOURCE_ROOT / "webui" / "templates" / "dashboard.html").read_text(encoding="utf-8")
        self.assertIn('name="douyin_network_mode"', dashboard)
        self.assertIn('name="douyin_proxy_url"', dashboard)
        browser = (SOURCE_ROOT / "core" / "browser.py").read_text(encoding="utf-8")
        self.assertIn("--no-proxy-server", browser)
        self.assertIn("SPARKFLOW_DOUYIN_NETWORK_MODE", browser)

    def test_login_desktop_resource_controls_are_configured(self):
        compose = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8")
        start_script = (SOURCE_ROOT / "scripts" / "start_login_desktop.sh").read_text(encoding="utf-8")
        server = (SOURCE_ROOT / "login_desktop_server.py").read_text(encoding="utf-8")
        self.assertIn("cpus: ${LOGIN_DESKTOP_CPUS:-0.8}", compose)
        self.assertIn("mem_limit: ${LOGIN_DESKTOP_MEMORY_LIMIT:-1200m}", compose)
        self.assertIn("pids_limit: ${LOGIN_DESKTOP_PIDS_LIMIT:-256}", compose)
        self.assertIn("-nap -wait 50 -defer 50", start_script)
        self.assertIn("--window-size=1600,1000", server)
        self.assertIn("start_idle_monitor", server)
        self.assertIn("schedule_stop_after_export", server)
        self.assertIn('reduced_motion="reduce"', server)
        self.assertIn('"/creator-micro/" in page.url', server)
        self.assertIn('"qr_ready": qr_ready', server)

    def test_login_desktop_uses_fastapi_lifespan(self):
        server = (SOURCE_ROOT / "login_desktop_server.py").read_text(encoding="utf-8")
        self.assertIn("lifespan=lifespan", server)
        self.assertNotIn("@app.on_event", server)

    def test_login_desktop_exposes_cropped_qr_endpoint(self):
        server = (SOURCE_ROOT / "login_desktop_server.py").read_text(encoding="utf-8")
        self.assertIn('@app.get("/qr")', server)
        self.assertIn('@app.post("/refresh-qr")', server)
        self.assertIn("qr_refresh=", server)
        self.assertIn('img[class*="qrcode"]', server)
        self.assertIn('qrcode_expired', server)
        self.assertIn('Cache-Control": "no-store, max-age=0', server)

    def test_legacy_unused_entrypoints_are_removed(self):
        self.assertFalse((SOURCE_ROOT / "webui" / "login_sessions.py").exists())
        self.assertFalse((SOURCE_ROOT / "relogin_worker.py").exists())
        self.assertFalse((SOURCE_ROOT / "docker-compose.example.yml").exists())


if __name__ == "__main__":
    unittest.main()