Give pin materialize the CI git channel and a private database

Why: test-pg has no SSH known_hosts, and the shared agent_test database already holds another instance identity.
Evidence: SELF python3 -m unittest tests.test_ci_credentials.CheckoutTests.test_rdm_pin_fetch_is_three_shas_on_the_allowlist tests.integration.lifecycle.test_lifecycle_real (1 pass, 24 skip).
Impact: lifecycle children forward GIT_CONFIG_* only; serve stays env -i. Login and rdm.ping stay NOT_RUN.
Refs: .task/feature-svc-d3-real-scenarios.md
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-10-13 04:03:55 +00:00
co-authored by Cursor
parent 678590b1dc
commit 7331323df4
4 changed files with 40 additions and 3 deletions
@@ -15,8 +15,14 @@ import time
import unittest
import urllib.error
import urllib.request
from urllib.parse import urlparse, urlunparse
from pathlib import Path
from tests.test_svc_realcombo import ci_git_env
from urllib.parse import urlparse, urlunparse
from tests.test_svc_realcombo import ci_git_env
SOURCE = Path(__file__).resolve().parents[3]
PIN = Path(__file__).resolve().parent / "rdm.pin"
@@ -55,6 +61,7 @@ class LifecycleRealTests(tsr.SvcAutoRecoverTests):
self.pin = load_pin()
self.env["SVC_RDM_SRC_DIR"] = SRC
self.env["RDM_DISABLE_EXEC_TOOLS"] = "1"
self.env.update(ci_git_env())
self.env.pop("AGENT_SYSTEM_DSN", None)
self.env.pop("AGENT_SYSTEM_DATA", None)
@@ -516,6 +523,7 @@ class LifecycleRealTests(tsr.SvcAutoRecoverTests):
dsn = os.environ.get("AGENT_PG_TEST_DSN", "")
if not dsn:
self.skipTest("PG 跨实例 NOT_RUN:未设置 AGENT_PG_TEST_DSN")
dsn = self.private_pg_dsn(dsn)
port = self.free_port()
self.seed_established("pga", port)
self.svc("pga", "stop")
@@ -572,6 +580,23 @@ class LifecycleRealTests(tsr.SvcAutoRecoverTests):
self.assertNotIn(sentinel, text, path)
self.svc("cred", "stop")
def private_pg_dsn(self, dsn):
"""共享的 AGENT_PG_TEST_DSN 已被其他用例写入身份。本用例单独建库。"""
parsed = urlparse(dsn)
name = "d3pg_" + hashlib.sha256(os.urandom(8)).hexdigest()[:12]
created = subprocess.run(
["psql", dsn, "-v", "ON_ERROR_STOP=1", "-c",
'CREATE DATABASE "%s"' % name],
text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=30)
self.assertEqual(created.returncode, 0, created.stdout.replace(dsn, "<dsn>"))
self.addCleanup(self.drop_private_pg, dsn, name)
return urlunparse(parsed._replace(path="/" + name))
def drop_private_pg(self, dsn, name):
subprocess.run(
["psql", dsn, "-c", 'DROP DATABASE IF EXISTS "%s" WITH (FORCE)' % name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30)
def event_count(self, name):
db = self.runtime(name) / "data" / "rdm.db"
con = sqlite3.connect("file:%s?mode=ro" % db, uri=True)
+5
View File
@@ -286,6 +286,11 @@ class CheckoutTests(unittest.TestCase):
digests = (work / "rdm" / "pin-build-digests.txt").read_text()
for sha in shas:
self.assertIn(sha, digests)
gate = WORKFLOW.read_text().split(
"Real-combination lifecycle gate", 1)[1].split("\n svc-realcombo:", 1)[0]
self.assertIn('export SVC_CI_GIT_TOKEN="$SUBMODULE_TOKEN"', gate)
self.assertNotRegex(gate, r"(?m)^\s*source\s")
self.assertNotIn(">>", gate)
if __name__ == "__main__":