DataOps-platform/tests/test_phase3_wp02_enterprise_identity_contract.py

from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def test_enterprise_identity_migration_is_normalized_forward_only():
    content = (ROOT / "migrations/versions/20260802_470_enterprise_identity.py").read_text()
    assert 'revision = "20260802_470"' in content
    assert 'down_revision = "20260802_460"' in content
    for table in (
        "identity_idp_config_versions", "identity_oidc_flows", "enterprise_identity_links",
        "identity_organization_nodes", "identity_directory_events", "identity_sessions",
        "identity_exchange_codes", "identity_emergency_requests", "identity_emergency_approvals",
        "identity_audit_events", "identity_directory_checkpoints", "identity_directory_conflicts",
    ):
        assert f"CREATE TABLE public.{table}" in content
    hardening = (ROOT / "migrations/versions/20260802_471_enterprise_identity_hardening.py").read_text()
    assert 'revision = "20260802_471"' in hardening
    assert 'down_revision = "20260802_470"' in hardening
    assert "^env:DATAOPS_OIDC_[A-Z0-9_]+$" in hardening
    assert "refresh_hash CHAR(64) NOT NULL UNIQUE" in content
    assert "emergency_request_uid UUID" in content
    assert "Forward-only migration" in content


def test_identity_api_uses_exact_public_routes_and_scoped_permissions():
    permissions = (ROOT / "app/core/system/permissions.py").read_text()
    api = (ROOT / "app/api/system/enterprise_identity.py").read_text()
    for permission in ("identity:read", "identity:manage", "identity:operate"):
        assert permission in permissions
    for path in ("providers", "authorize", "callback", "exchange", "refresh"):
        assert f'"/api/system/identity/{path}"' in permissions
    assert 'path.startswith("/api/system/auth/")' not in permissions
    for endpoint in ("/identity/providers", "/identity/authorize", "/identity/callback", "/identity/exchange",
                     "/identity/refresh", "/identity/logout", "/identity/idp-versions", "/identity/claims/dry-run",
                     "/identity/directory/delta", "/identity/sessions", "/identity/emergency/requests", "/identity/audit"):
        assert endpoint in api
    assert "OidcTransport" in api
    assert "client_secret" in (ROOT / "app/core/system/oidc.py").read_text()
    assert "safe_detail" in api
    assert "access_token" not in api.split("response = redirect", 1)[1].split("return response", 1)[0]
    repository = (ROOT / "app/core/system/identity_repository.py").read_text()
    assert "WHERE refresh_hash=:refresh_hash AND status='active'" in repository
    assert "RETURNING family_uid::text" in repository
    assert "provider_uid=CAST(:provider AS uuid) AND enterprise_subject=:subject" in repository
    assert "provider_disabled" in api
    for audit_event in ("idp_dry_run", "idp_connectivity", "claims_dry_run", "identity_refresh",
                        "identity_logout", "identity_session_revoke", "identity_session_risk_terminate",
                        "identity_session_revoke_all", "emergency_request"):
        assert f'"{audit_event}"' in api
    assert "pg_advisory_xact_lock" in repository
    assert "SELECT id FROM public.users WHERE id=CAST(:user AS uuid) FOR UPDATE" in repository
    assert "put_organization_node" in repository and "apply_directory_event_atomic" in repository
    assert 'visible = ("uid", "user_uid", "identity_source", "status", "created_at", "last_seen_at", "expires_at")' in api
    assert '"family_uid"' not in api.split("def identity_sessions():", 1)[1].split("def identity_session_revoke", 1)[0]
    auth = (ROOT / "app/core/system/auth.py").read_text()
    assert 'if active_provider and not claims.get("sid")' in auth
    assert "enterprise identity state unavailable; local login denied" in auth
    assert "provider_configuration_changed" in api
    assert "commit=False" in api and "insert_exchange_code" in api


def test_frontend_has_sso_callback_secure_storage_single_refresh_and_management_surface():
    auth = (ROOT / "frontend/src/utils/auth.js").read_text()
    request = (ROOT / "frontend/src/utils/request.js").read_text()
    routes = (ROOT / "frontend/src/router/defaultRoute.js").read_text()
    menu = (ROOT / "frontend/src/router/routes.js").read_text()
    callback = (ROOT / "frontend/src/views/login/callback.vue").read_text()
    management = (ROOT / "frontend/src/views/systemManage/enterpriseIdentity/index.vue").read_text()
    user_store = (ROOT / "frontend/src/store/modules/user.js").read_text()
    assert "sessionStorage.getItem('access_token')" in auth
    assert "localStorage.removeItem('access_token')" in auth
    assert "delete profile.token" in user_store and "delete userInfo.token" in user_store
    assert "let refreshPromise = null" in request
    assert "__identityRetry" in request
    assert "'/login/callback'" in routes and "noLogin: true" in routes
    assert "history.replaceState" in callback and "exchangeIdentityCode" in callback
    assert "/systemManage/enterprise-identity" in menu and "identity:manage" in menu
    assert "待企业 IdP UAT" in management
    for operation in ("createIdpVersion", "runIdpAction", "claimsDryRun", "syncDirectoryDelta",
                      "requestEmergencyAccess", "runEmergencyAction"):
        assert operation in management


def test_phase3_identity_docs_and_status_are_fail_closed():
    delivery = (ROOT / "docs/phase3/P3_WP02_ENTERPRISE_IDENTITY.md").read_text()
    evidence = (ROOT / "docs/validation/P3_WP02_ENTERPRISE_IDENTITY_EVIDENCE.md").read_text()
    required = (ROOT / "docs/phase3/P3_WP00_REQUIREMENTS.json").read_text()
    status = "ENGINEERING_COMPLETE_ENTERPRISE_IDP_UAT_BLOCKED"
    assert status in delivery and status in evidence and status in required
    assert "TBD_EXTERNAL" in delivery and "TBD_EXTERNAL" in evidence