metis-desk/scripts/resource-scope-test.py
2026-10-25 01:19:17 +02:00

98 lines
7.2 KiB
Python

#!/usr/bin/env python3
"""Ensaya Resources Scope exclusivamente en MariaDB desechable; no carga .env."""
import argparse
import json
import os
from pathlib import Path
import secrets
import subprocess
import tempfile
import time
import uuid
ROOT = Path(__file__).resolve().parents[1]
IMAGE = "mariadb:11.8@sha256:6422478cb8e159f080fb1d8ccf65101e26fe51385787fde7d16c3b165a331f15"
def command(args, **kwargs):
return subprocess.run(args, check=True, capture_output=True, text=True, timeout=60, **kwargs)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--preflight", action="store_true")
parser.add_argument("--coincidences", action="store_true", help="Ejecuta R2.6 de solo lectura en MariaDB desechable")
parser.add_argument("--profiles", action="store_true", help="Ejecuta R2.5 sobre MariaDB desechable")
parser.add_argument("--software", action="store_true", help="Ejecuta R2.4 sobre MariaDB desechable")
parser.add_argument("--catalog", action="store_true", help="Ejecuta R2.3 sobre MariaDB desechable")
parser.add_argument("--security", action="store_true", help="Ejecuta R2.2 sobre MariaDB desechable")
parser.add_argument("--regressions", action="store_true", help="Ejecuta las regresiones HTTP/MariaDB sobre el esquema expandido")
parser.add_argument("--monorepo", action="store_true", help="Añade tests de paquetes sin cargar .env")
options = parser.parse_args()
artifacts = ROOT / (".artifacts/resources-global-scope/r2-6" if options.coincidences else ".artifacts/resources-global-scope/r2-5" if options.profiles else ".artifacts/resources-global-scope/r2-4" if options.software else (".artifacts/resources-global-scope/r2-3" if options.catalog else (".artifacts/resources-global-scope/r2-2" if options.security else ".artifacts/resources-global-scope/r2-1b")))
artifacts.mkdir(parents=True, exist_ok=True)
container = "metis-scope-test-" + uuid.uuid4().hex[:12]
password = secrets.token_urlsafe(32)
created = False
with tempfile.TemporaryDirectory(prefix="metis-scope-") as temporary:
private = Path(temporary) / "database.env"
private.touch(mode=0o600)
private.write_text("MARIADB_DATABASE=metis_desk_test\nMARIADB_USER=scope_test\nMARIADB_PASSWORD=" + password + "\nMARIADB_ROOT_PASSWORD=" + secrets.token_urlsafe(32) + "\n")
try:
command(["docker", "run", "--detach", "--rm", "--name=" + container,
"--label=metis.resource-scope-test=true", "--publish=127.0.0.1::3306",
"--env-file=" + str(private), IMAGE])
created = True
private.unlink()
deadline = time.monotonic() + 120
while subprocess.run(["docker", "exec", container, "healthcheck.sh", "--connect", "--innodb_initialized"], capture_output=True, timeout=15).returncode:
if time.monotonic() >= deadline:
raise RuntimeError("MariaDB desechable no está preparada tras 120 segundos.")
time.sleep(0.25)
bindings = json.loads(command(["docker", "inspect", "--format={{json .NetworkSettings.Ports}}", container]).stdout)
environment = {**os.environ, "DB_HOST": "127.0.0.1", "DB_PORT": bindings["3306/tcp"][0]["HostPort"],
"DB_NAME": "metis_desk_test", "DB_USER": "scope_test", "DB_PASSWORD": password,
"NODE_ENV": "test", "SESSION_SECRET": secrets.token_urlsafe(48), "APP_ORIGIN": "http://localhost:5173",
"METIS_RESOURCE_SCOPE_PRIVATE": temporary, "METIS_RESOURCE_SCOPE_DISPOSABLE": container, "METIS_RESOURCE_SCOPE_ARTIFACTS": str(artifacts)}
target = "apps/server/src/application-coincidences.integration.test.ts" if options.coincidences else "apps/server/src/global-profile-applications.integration.test.ts" if options.profiles else "apps/server/src/global-software.integration.test.ts" if options.software else "apps/server/src/resource-scope.preflight.ts" if options.preflight else ("apps/server/src/global-applications.integration.test.ts" if options.catalog else ("apps/server/src/resource-authorization.integration.test.ts" if options.security else "apps/server/src/resource-scope.integration.test.ts"))
args = ["node", "--import", "tsx"]
if not options.preflight:
args += ["--test", "--test-isolation=none"]
with (artifacts / ("preflight.log" if options.preflight else "integration.log")).open("w") as log:
result = subprocess.run([*args, target], cwd=ROOT, env=environment, stdout=log, stderr=subprocess.STDOUT, timeout=600)
print("Evidencias: " + str(artifacts.relative_to(ROOT)), flush=True)
if result.returncode:
raise SystemExit(result.returncode)
results = {"integration": result.returncode}
if options.regressions and not options.preflight:
with (artifacts / "migration-cli.log").open("w") as log:
migrated = command(["node", "--import", "tsx", "apps/server/src/migrate-resource-scope.ts", "--apply", "--writes-paused"], cwd=ROOT, env=environment)
log.write(migrated.stdout)
suites = ["app.test.ts", "infrastructure.http.test.ts", "config/env-schema.test.ts",
"infrastructure/http/http-error-diagnostic.test.ts", "infrastructure/http/relationship-read-access.test.ts",
"network-shares.integration.test.ts", "directory-domains.integration.test.ts",
"smb-access.integration.test.ts", "asset-directory-memberships.integration.test.ts",
"typed-relationships.integration.test.ts", "network-share-publications.integration.test.ts",
"template-propagation.integration.test.ts"]
with (artifacts / "regressions.log").open("w") as log:
result = subprocess.run(["node", "--import", "tsx", "--test", "--test-isolation=none", "--test-concurrency=1", *["apps/server/src/" + suite for suite in suites]], cwd=ROOT, env=environment, stdout=log, stderr=subprocess.STDOUT, timeout=600)
results["regressions"] = result.returncode
if options.monorepo:
with (artifacts / "legacy-server.log").open("w") as log:
result = subprocess.run(["node", "--import", "tsx", "--test", "--test-isolation=none", "apps/server/src/identity.test.ts"], cwd=ROOT, env=environment, stdout=log, stderr=subprocess.STDOUT, timeout=600)
results["legacyServer"] = result.returncode
with (artifacts / "monorepo-tests.log").open("w") as log:
result = subprocess.run(["pnpm", "exec", "turbo", "run", "test", "--filter=!@metis/server"], cwd=ROOT, env=environment, stdout=log, stderr=subprocess.STDOUT, timeout=600)
results["monorepo"] = result.returncode
(artifacts / "commands.json").write_text(json.dumps(results, indent=2) + "\n")
if any(results.values()):
raise SystemExit(1)
finally:
if created:
command(["docker", "rm", "-f", "-v", container])
print("MariaDB desechable retirada.", flush=True)
if __name__ == "__main__":
main()