From 873dca205129f123d5ea9dd768bfa1c2e5452830 Mon Sep 17 00:00:00 2001
From: MB-X Bilibili Pipeline <mbx-bili-pipeline@localhost>
Date: Sat, 05 Sep 2026 22:27:58 +0800
Subject: [PATCH] chore(project-info): archive Bilibili dynamic 66e47c0ffa2d
---
dev/project-dev/test/test_bili_dynamic_collector.py | 12 +++++++++---
1 files changed, 9 insertions(+), 3 deletions(-)
diff --git a/dev/project-dev/test/test_bili_dynamic_collector.py b/dev/project-dev/test/test_bili_dynamic_collector.py
index 31e2505..b33d6b9 100644
--- a/dev/project-dev/test/test_bili_dynamic_collector.py
+++ b/dev/project-dev/test/test_bili_dynamic_collector.py
@@ -91,6 +91,12 @@
text = stream.getvalue()
return code, json.loads(text), text
+ def assert_lock_quiescent(self) -> None:
+ lock = self.state_dir / ".collector.lock"
+ self.assertTrue(lock.is_file())
+ self.assertEqual(b"\0", lock.read_bytes())
+ self.assertFalse((self.state_dir / ".collector.lock.owner.json").exists())
+
def queue_video(self) -> dict[str, object]:
code, result = self.run_check()
self.assertEqual(0, code)
@@ -462,7 +468,7 @@
self.assertEqual("E_STATUS", payload["error_code"])
self.assertEqual(before, after)
self.assertEqual([], list(self.video_dir.iterdir()))
- self.assertFalse((self.state_dir / ".collector.lock").exists())
+ self.assert_lock_quiescent()
def test_manifest_secret_gate_is_shared_by_all_commands(self) -> None:
queued = self.queue_video()
@@ -556,7 +562,7 @@
if isinstance(value, str) and value.strip():
self.assertNotIn(value, stdout)
self.assertEqual([], list(self.video_dir.iterdir()))
- self.assertFalse((self.state_dir / ".collector.lock").exists())
+ self.assert_lock_quiescent()
def test_handoff_rejects_control_or_invalid_manifest_scalars_without_writes(self) -> None:
queued = self.queue_video()
@@ -615,7 +621,7 @@
self.assertNotIn(value, stdout)
self.assertEqual(before, after)
self.assertEqual(source_bytes, target.read_bytes())
- self.assertFalse((self.state_dir / ".collector.lock").exists())
+ self.assert_lock_quiescent()
def test_native_handoff_route_is_required_and_validated_before_state_write(self) -> None:
base = json.loads(self.config_path.read_text(encoding="utf-8"))
--
Gitblit v1.9.3