metis-desk/scripts/visual-audit.test.py

#!/usr/bin/env python3
"""Regresión del aislamiento de las sesiones del lanzador, sin Docker ni credenciales."""

import json
import os
from pathlib import Path
import runpy
import subprocess
import tempfile
import tomllib
import unittest
import uuid

ROOT = Path(__file__).resolve().parent.parent


class SessionLauncherTests(unittest.TestCase):
    def setUp(self):
        artifacts = ROOT / ".artifacts/playwright"
        artifacts.mkdir(parents=True, exist_ok=True)
        self.directory = tempfile.TemporaryDirectory(prefix="session-guard-", dir=artifacts)
        self.root = Path(self.directory.name)
        self.executable = self.root / "codex-probe"
        self.executable.write_text("#!/usr/bin/env python3\nimport json,sys\nprint(json.dumps({'args':sys.argv[1:],'stdin':sys.stdin.read()}))\n")
        self.executable.chmod(0o700)

    def tearDown(self):
        self.directory.cleanup()

    def state(self, domain="localhost", mode=0o600):
        path = self.root / "session.json"
        path.write_text(json.dumps({"cookies": [{"name": "metis.sid", "domain": domain,
                                                  "httpOnly": True, "value": "placeholder-not-a-session"}]}))
        path.chmod(mode)
        return path

    def launch(self, state=None):
        environment = {**os.environ, "METIS_CODEX_BIN": str(self.executable)}
        environment.pop("METIS_PLAYWRIGHT_STORAGE_STATE", None)
        if state is not None:
            environment["METIS_PLAYWRIGHT_STORAGE_STATE"] = str(state)
        return subprocess.run(["bash", str(ROOT / "scripts/codex-playwright-wsl.sh"), "mcp", "list", "--json"],
                              input="stdin-check", text=True, capture_output=True, env=environment,
                              cwd=self.root, timeout=15, check=False)

    def args(self, result):
        self.assertEqual(result.returncode, 0, result.stderr)
        payload = json.loads(result.stdout)
        self.assertEqual(payload["stdin"], "stdin-check")
        argument = next(item for item in payload["args"] if item.startswith("mcp_servers.playwright.args="))
        return tomllib.loads(argument)["mcp_servers"]["playwright"]["args"]

    def test_default_has_no_session(self):
        self.assertFalse(any("storage-state" in item or "type=bind" in item for item in self.args(self.launch())))

    def test_private_session_is_readonly_and_not_in_arguments(self):
        state = self.state()
        args = self.args(self.launch(state.relative_to(ROOT)))
        self.assertTrue(any("readonly" in item and str(state) in item for item in args))
        self.assertEqual(args.count("--storage-state=/tmp/metis-audit-state.json"), 1)
        self.assertNotIn("placeholder-not-a-session", " ".join(args))

    def test_public_permissions_are_rejected(self):
        result = self.launch(self.state(mode=0o644))
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("privado", result.stderr)

    def test_remote_cookie_is_rejected(self):
        result = self.launch(self.state(domain="production.example.test"))
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("locales", result.stderr)

    def test_external_file_is_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "session.json"
            path.write_text("{}")
            path.chmod(0o600)
            result = self.launch(path)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("privado", result.stderr)

    def test_missing_session_is_rejected(self):
        result = self.launch(self.root / "missing.json")
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("ausente", result.stderr)


class ResourceFixtureTests(unittest.TestCase):
    def setUp(self):
        self.prepare = runpy.run_path(str(ROOT / "scripts/visual-audit.py"))["prepare_resources"]
        keys = ["template:" + name for name in ["application", "shared_folder", "software_license", "certificate", "windows_domain"]]
        keys += ["field:" + name for name in ["provider", "url", "description", "unc", "server", "software_product", "software_edition", "license_model", "license_metric", "license_quantity", "auto_renew", "reference", "expires", "renews", "holder", "issuer", "domain_dns", "domain_netbios"]]
        self.catalog = [{"id": str(uuid.uuid4()), "builtinKey": key} for key in keys]
        self.resources = []
        self.posts = []
        self.company = str(uuid.uuid4())

    def request(self, path, body=None, expected=200):
        if path == "/resource-catalog/" + self.company:
            return {"items": self.catalog}
        self.assertEqual(path, "/resources/" + self.company)
        if body is None:
            return {"resources": self.resources}
        self.assertEqual(expected, 201)
        resource = {"id": str(uuid.uuid4()), **body}
        self.posts.append(body)
        self.resources.append(resource)
        return {"resource": resource}

    def test_resources_are_repeatable_without_duplicate_writes(self):
        first = self.prepare(self, self.company)
        count = len(self.posts)
        self.assertGreater(count, 10)
        self.assertEqual(self.prepare(self, self.company), first)
        self.assertEqual(len(self.posts), count)
        self.assertEqual({row["kind"] for row in first}, {"application", "shared_folder", "software_license", "certificate", "windows_domain"})

    def test_missing_catalog_fails_before_creating_resources(self):
        self.catalog = [item for item in self.catalog if item["builtinKey"] != "field:provider"]
        with self.assertRaisesRegex(RuntimeError, "Falta un campo"):
            self.prepare(self, self.company)
        self.assertEqual(self.posts, [])

    def test_values_respect_limits_and_preserve_zero_and_false(self):
        self.prepare(self, self.company)
        fields = {item["builtinKey"]: item["id"] for item in self.catalog}
        license_values = next(item["values"] for item in self.resources if item["values"]["name"].startswith("AUDIT-LICENSE"))
        self.assertEqual(license_values["characteristics"][fields["field:license_quantity"]], 0)
        self.assertIs(license_values["characteristics"][fields["field:auto_renew"]], False)
        self.assertEqual({item["values"]["active"] for item in self.resources}, {True, False})
        self.assertTrue(all(len(item["values"]["name"]) <= 100 and len(item["values"]["notes"]) <= 10000 for item in self.resources))


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