Creates and initializes a core lifecycle instance with a temporary database.
(tmp_path_factory)
| 23 | |
| 24 | @pytest_asyncio.fixture(scope="module") |
| 25 | async def core_lifecycle_td(tmp_path_factory): |
| 26 | """Creates and initializes a core lifecycle instance with a temporary database.""" |
| 27 | tmp_db_path = tmp_path_factory.mktemp("data") / "test_data_kb.db" |
| 28 | db = SQLiteDatabase(str(tmp_db_path)) |
| 29 | log_broker = LogBroker() |
| 30 | core_lifecycle = AstrBotCoreLifecycle(log_broker, db) |
| 31 | await core_lifecycle.initialize() |
| 32 | |
| 33 | # Mock kb_manager and kb_helper |
| 34 | kb_manager = MagicMock() |
| 35 | kb_helper = AsyncMock(spec=KBHelper) |
| 36 | |
| 37 | # Configure get_kb to be an async mock that returns kb_helper |
| 38 | kb_manager.get_kb = AsyncMock(return_value=kb_helper) |
| 39 | |
| 40 | # Mock upload_document return value |
| 41 | mock_doc = KBDocument( |
| 42 | doc_id="test_doc_id", |
| 43 | kb_id="test_kb_id", |
| 44 | doc_name="test_file.txt", |
| 45 | file_type="txt", |
| 46 | file_size=100, |
| 47 | file_path="", |
| 48 | chunk_count=2, |
| 49 | media_count=0, |
| 50 | ) |
| 51 | kb_helper.upload_document.return_value = mock_doc |
| 52 | |
| 53 | # kb_manager.get_kb.return_value = kb_helper # Removed this line as it's handled above |
| 54 | core_lifecycle.kb_manager = kb_manager |
| 55 | generated_password = getattr( |
| 56 | core_lifecycle.astrbot_config, |
| 57 | "_generated_dashboard_password", |
| 58 | None, |
| 59 | ) |
| 60 | dashboard_password = generated_password or _TEST_DASHBOARD_PASSWORD |
| 61 | if not generated_password: |
| 62 | core_lifecycle.astrbot_config["dashboard"]["pbkdf2_password"] = ( |
| 63 | hash_dashboard_password(dashboard_password) |
| 64 | ) |
| 65 | core_lifecycle.astrbot_config["dashboard"]["password"] = ( |
| 66 | hash_md5_dashboard_password(dashboard_password) |
| 67 | ) |
| 68 | object.__setattr__( |
| 69 | core_lifecycle, |
| 70 | "_dashboard_plain_password", |
| 71 | dashboard_password, |
| 72 | ) |
| 73 | |
| 74 | try: |
| 75 | yield core_lifecycle |
| 76 | finally: |
| 77 | try: |
| 78 | _stop_res = core_lifecycle.stop() |
| 79 | if asyncio.iscoroutine(_stop_res): |
| 80 | await _stop_res |
| 81 | except Exception: |
| 82 | pass |
nothing calls this directly
no test coverage detected