zapretgui/tests/test_blockcheck_worker_architecture.py

from __future__ import annotations

import inspect
import unittest

from app.feature_facades.blockcheck import BlockcheckFeature
import blockcheck.commands as blockcheck_commands
import blockcheck.page_runtime as blockcheck_page_runtime
import blockcheck.strategy_scan_worker as strategy_scan_worker
import blockcheck.worker as blockcheck_worker
import blockcheck.strategy_apply_worker as strategy_apply_worker
import blockcheck.workers as blockcheck_workers


class BlockcheckWorkerArchitectureTests(unittest.TestCase):
    def test_scan_finalize_worker_receives_feature_action_callable(self) -> None:
        feature_source = inspect.getsource(BlockcheckFeature.create_strategy_scan_finalize_worker)
        worker_source = inspect.getsource(blockcheck_workers.StrategyScanFinalizeWorker)

        self.assertNotIn("blockcheck_feature=self", feature_source)
        self.assertNotIn("self._blockcheck", worker_source)
        self.assertIn("finalize_scan_report=self.finalize_scan_report", feature_source)
        self.assertIn("_finalize_scan_report", worker_source)
        self.assertFalse(hasattr(blockcheck_workers, "StrategyScanResumeSaveWorker"))
        self.assertNotIn("blockcheck_public.", worker_source)
        self.assertNotIn("import blockcheck.public", worker_source)

    def test_strategy_apply_worker_uses_apply_callable_not_feature_object(self) -> None:
        feature_source = inspect.getsource(BlockcheckFeature.create_strategy_apply_worker)
        worker_init_source = inspect.getsource(strategy_apply_worker.StrategyApplyWorker.__init__)
        worker_run_source = inspect.getsource(strategy_apply_worker.StrategyApplyWorker.run)

        self.assertNotIn("blockcheck_feature=self", feature_source)
        self.assertIn("apply_strategy", worker_init_source)
        self.assertIn("self._apply_strategy", worker_init_source)
        self.assertNotIn("self._blockcheck", worker_init_source)
        self.assertIn("self._apply_strategy(", worker_run_source)
        self.assertNotIn("self._blockcheck.apply_strategy", worker_run_source)

    def test_blockcheck_run_worker_receives_log_actions_as_callables(self) -> None:
        feature_source = inspect.getsource(BlockcheckFeature.create_blockcheck_worker)
        worker_source = inspect.getsource(blockcheck_worker.BlockcheckWorker)

        # У проверки один файл — отчёт .json: рабочий поток получает, куда его писать и чем.
        self.assertIn("report_path=self.make_blockcheck_report_path", feature_source)
        self.assertIn("save_report=self.save_blockcheck_report", feature_source)
        self.assertIn("_report_path", worker_source)
        self.assertIn("_save_report", worker_source)
        # Текстовый журнал проверки больше не ведётся.
        self.assertNotIn("append_run_log", worker_source)
        self.assertNotIn("blockcheck.commands", worker_source)

    def test_blockcheck_run_log_file_io_lives_in_session_registry_not_page_runtime(self) -> None:
        commands_source = inspect.getsource(blockcheck_commands)
        page_runtime_source = inspect.getsource(blockcheck_page_runtime)

        self.assertIn("def save_blockcheck_report", commands_source)
        self.assertNotIn("def append_blockcheck_run_log", commands_source)
        self.assertTrue(blockcheck_commands.make_blockcheck_report_path("full").endswith("_full.json"))

        self.assertNotIn("with open(", page_runtime_source)
        self.assertNotIn("os.makedirs(", page_runtime_source)

    def test_strategy_scan_worker_receives_log_actions_as_callables(self) -> None:
        feature_source = inspect.getsource(BlockcheckFeature.create_strategy_scan_worker)
        worker_source = inspect.getsource(strategy_scan_worker.StrategyScanWorker)

        self.assertIn("start_run_log=self.start_strategy_scan_run_log", feature_source)
        self.assertIn("append_run_log=self.append_strategy_scan_run_log", feature_source)
        self.assertIn("close_run_log=self.close_strategy_scan_run_log", feature_source)
        self.assertIn("_start_run_log_action", worker_source)
        self.assertIn("_append_run_log_action", worker_source)
        self.assertIn("_close_run_log_action", worker_source)
        self.assertNotIn("blockcheck.commands", worker_source)

    def test_strategy_scan_worker_exposes_finished_signal_for_shared_runtime(self) -> None:
        worker_source = inspect.getsource(strategy_scan_worker.StrategyScanWorker)

        self.assertTrue(hasattr(strategy_scan_worker.StrategyScanWorker, "finished"))
        self.assertIn("finished = pyqtSignal(object)", worker_source)
        self.assertIn("self.finished.emit(report)", worker_source)
        run_source = inspect.getsource(strategy_scan_worker.StrategyScanWorker.run)
        self.assertLess(run_source.index("_close_run_log_action"), run_source.index("self.finished.emit(report)"))

    def test_blockcheck_run_workers_leave_thread_ownership_to_shared_runtime(self) -> None:
        worker_sources = "\n".join(
            (
                inspect.getsource(blockcheck_worker.BlockcheckWorker),
                inspect.getsource(strategy_scan_worker.StrategyScanWorker),
            )
        )

        self.assertIn("def run(self)", worker_sources)
        self.assertNotIn("threading.Thread", worker_sources)
        self.assertNotIn("_bg_thread", worker_sources)
        self.assertNotIn("def start(self)", worker_sources)

    def test_strategy_scan_worker_receives_runtime_shutdown_callable_not_full_runtime_feature(self) -> None:
        from blockcheck.strategy_search.environment import RealEnvironment

        feature_source = inspect.getsource(BlockcheckFeature.create_strategy_scan_worker)
        worker_signature = inspect.signature(strategy_scan_worker.StrategyScanWorker.__init__)
        environment_signature = inspect.signature(RealEnvironment.__init__)
        worker_source = inspect.getsource(strategy_scan_worker.StrategyScanWorker)
        environment_source = inspect.getsource(RealEnvironment)

        self.assertNotIn("runtime_feature", worker_signature.parameters)
        self.assertNotIn("runtime_feature", environment_signature.parameters)
        self.assertNotIn("_runtime_feature", worker_source)
        self.assertNotIn("_runtime_feature", environment_source)
        self.assertIn("shutdown_sync", worker_signature.parameters)
        self.assertIn("shutdown_sync", environment_signature.parameters)
        self.assertIn("shutdown_sync=shutdown_sync", feature_source)
        # Старая фабрика без реестра фейков удалена: подбор без --blob= работал бы как pass.
        self.assertFalse(hasattr(blockcheck_commands, "create_strategy_scan_worker"))


if __name__ == "__main__":
    unittest.main()