hololake-system-architecture/deployment/enterprise-domain-bootstrap/enterprise-lighthouse/enterprise_identity_service.test.py

import importlib.util
import json
import os
import tempfile
import unittest
import urllib.error
from pathlib import Path
from unittest import mock

ROOT = Path(__file__).parent
REGISTRY = ROOT / "registry" / "enterprise-identity-registry.json"
os.environ["GH_ENTERPRISE_IDENTITY_REGISTRY"] = str(REGISTRY)
os.environ["GH_ENTERPRISE_RECEIPT_KEY"] = "test-only-key-that-is-longer-than-32-bytes"
spec = importlib.util.spec_from_file_location("enterprise_identity_service", ROOT / "enterprise_identity_service.py")
service = importlib.util.module_from_spec(spec)
spec.loader.exec_module(service)


class EnterpriseIdentityTests(unittest.TestCase):
    def test_age_is_species_and_never_an_individual_identity(self):
        registry = service.load_registry()
        self.assertEqual(registry["persona_identity_governance"]["species"], "AGE")
        self.assertFalse(registry["persona_identity_governance"]["age_is_individual_number_namespace"])
        for human in registry["humans"]:
            for persona in human["personas"]:
                self.assertEqual(persona["species"], "AGE")
                self.assertFalse(persona["current_persona_identity"].startswith("AGE-"))

    def test_five_humans_route_to_five_private_work_repositories(self):
        registry = service.load_registry()
        self.assertEqual(len(registry["humans"]), 5)
        self.assertEqual(len({item["repository"] for item in registry["humans"]}), 5)
        self.assertTrue(all(item["repository"].split("/")[0] == item["username"] for item in registry["humans"]))
        self.assertTrue(all(service.public_projection(registry, item)["work_entry"]["domain"] == "ZERO_SENSE_DOMAIN" for item in registry["humans"]))
        self.assertEqual(service.find_human(registry, "TCS-GL-0007∞")["username"], "feimao")
        self.assertIsNone(service.find_human(registry, "TCS-GL-9999∞"))

    def test_credentials_are_parsed_but_never_part_of_a_receipt(self):
        encoded = service.base64.b64encode(b"feimao:temporary-secret").decode()
        self.assertEqual(service.parse_basic(f"Basic {encoded}"), ("feimao", "temporary-secret"))
        receipt = service.signed_receipt({"receipt_id":"R1","human_number":"TCS-GL-0007∞","username":"feimao"})
        self.assertNotIn("password", json.dumps(receipt).lower())
        self.assertNotIn("temporary-secret", json.dumps(receipt))

    def test_database_separates_relationship_and_responsibility_receipts(self):
        with tempfile.TemporaryDirectory() as temp:
            old = service.DB_PATH
            service.DB_PATH = str(Path(temp) / "identity.sqlite3")
            try:
                db = service.database()
                tables = {row[0] for row in db.execute("select name from sqlite_master where type='table'")}
                self.assertIn("relationship_receipts", tables)
                self.assertIn("responsibility_receipts", tables)
                self.assertIn("credential_rotation_receipts", tables)
                db.close()
            finally:
                service.DB_PATH = old

    def test_password_rotation_source_uses_user_session_and_never_admin_token(self):
        source = (ROOT / "enterprise_identity_service.py").read_text()
        self.assertIn("rotate_forgejo_password", source)
        self.assertIn("/user/settings/change_password", source)
        self.assertNotIn("FORGEJO_ADMIN_TOKEN", source)

    def test_receipt_id_and_repository_path_are_stable_without_exposing_idempotency_key(self):
        first = service.stable_receipt_id("GH-RESP", "TCS-GL-0007∞", "responsibility-1234567890")
        second = service.stable_receipt_id("GH-RESP", "TCS-GL-0007∞", "responsibility-1234567890")
        self.assertEqual(first, second)
        self.assertRegex(first, r"^GH-RESP-[A-F0-9]{32}$")
        self.assertNotIn("1234567890", first)
        self.assertEqual(
            service.repository_receipt_path("responsibility", first),
            f".guanghu/receipts/responsibility/{first}.json",
        )

    def test_repository_projection_uses_the_humans_own_forgejo_authority(self):
        registry = service.load_registry()
        human = service.find_human(registry, "TCS-GL-0007∞")
        receipt_id = service.stable_receipt_id("GH-REL", human["human_number"], "relationship-1234567890")
        receipt = service.signed_receipt(
            {"receipt_id": receipt_id, "human_number": human["human_number"], "username": "feimao"}
        )

        class Response:
            status = 201
            def __enter__(self): return self
            def __exit__(self, *_): return False
            def read(self):
                return json.dumps({"commit": {"sha": "a" * 40}}).encode()

        with mock.patch.object(service.urllib.request, "urlopen", return_value=Response()) as opened:
            projection = service.project_receipt_to_repository(
                human, "feimao", "one-use-secret", "relationship", receipt
            )
        request = opened.call_args.args[0]
        self.assertEqual(projection["repository"], "feimao/guanghu-zero-sense-work")
        self.assertIn("/repos/feimao/guanghu-zero-sense-work/contents/", request.full_url)
        self.assertTrue(request.headers["Authorization"].startswith("Basic "))
        self.assertNotIn("one-use-secret", request.data.decode())

    def test_repository_projection_refuses_cross_owner_repository(self):
        human = {"repository": "juzi/guanghu-zero-sense-work"}
        with self.assertRaisesRegex(RuntimeError, "owner"):
            service.project_receipt_to_repository(
                human,
                "feimao",
                "secret",
                "relationship",
                {"receipt_id": "GH-REL-" + "A" * 32},
            )


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