From 9e244ddff6e5bf68624fa983c607a63135a39d43 Mon Sep 17 00:00:00 2001 From: Zohaib Hassnain <109234410+ZohaibHassan16@users.noreply.github.com> Date: Tue, 16 Jun 2026 04:43:45 +0500 Subject: [PATCH] Fix CLI demo blockers --- semantica/cli.py | 47 +++++- semantica/utils/progress_tracker.py | 177 +++++++++++---------- tests/test_cli_foundation.py | 115 ++++++++++++- tests/test_progress_tracker_regressions.py | 130 +++++++++++++++ 4 files changed, 375 insertions(+), 94 deletions(-) create mode 100644 tests/test_progress_tracker_regressions.py diff --git a/semantica/cli.py b/semantica/cli.py index e9235d2a..c9c362e6 100644 --- a/semantica/cli.py +++ b/semantica/cli.py @@ -220,6 +220,42 @@ def _get_framework(cli_ctx: CLIContext) -> "Semantica": return cli_ctx.framework +def _check_log_directory(cli_ctx: CLIContext) -> str: + """Verify the configured log directory is writable. + + The CLI may already hold an open file handle for ``semantica.log``. Probe + with a unique temporary file so Windows can check writability without + deleting the active log file. + """ + log_file = cli_ctx.config.get("logging.file", "semantica.log") + if not log_file: + return "console-only logging" + + log_path = Path(log_file).expanduser() + log_dir = log_path.parent + if not log_dir or str(log_dir) == ".": + log_dir = Path.cwd() + elif not log_dir.is_absolute(): + log_dir = Path.cwd() / log_dir + + probe_path: Optional[Path] = None + try: + probe_path = ( + log_dir + / f".semantica-doctor-{os.getpid()}-{time.time_ns()}.tmp" + ) + probe_path.write_text("ok", encoding="utf-8") + return f"{log_dir} writable" + finally: + if probe_path is not None: + try: + probe_path.unlink(missing_ok=True) + except OSError: + # Some locked-down Windows environments allow writes but deny + # unlinking. Logging can still proceed, so cleanup is best-effort. + pass + + def _run_build(cli_ctx: CLIContext, sources: Sequence[str]) -> None: """Thin wrapper around existing build orchestration flow. @@ -241,6 +277,10 @@ def _run_build(cli_ctx: CLIContext, sources: Sequence[str]) -> None: "At least one source is required. Use --source/-s one or more times." ) + if cli_ctx.dry_run_global: + _dry(cli_ctx, "build knowledge base", sources=list(sources)) + return + framework = _get_framework(cli_ctx) if cli_ctx.quiet or cli_ctx.json_output: result = framework.build_knowledge_base(sources=list(sources)) @@ -804,12 +844,7 @@ def doctor(cli_ctx: CLIContext, local_json: bool) -> None: else: checks.append(("Config file", "warn", "using defaults (no --config)", "run 'semantica init'")) - # Log directory writability - def _logdir() -> str: - p = Path.cwd() / "semantica.log" - p.touch(); p.unlink() - return "current directory writable" - checks.append(_check("Log directory", _logdir)) + checks.append(_check("Log directory", lambda: _check_log_directory(cli_ctx))) if _is_json(cli_ctx, local_json): _jecho([{"check": lbl, "status": st, "note": note, "hint": hint} diff --git a/semantica/utils/progress_tracker.py b/semantica/utils/progress_tracker.py index 481e0823..febfb4b9 100644 --- a/semantica/utils/progress_tracker.py +++ b/semantica/utils/progress_tracker.py @@ -54,6 +54,16 @@ DISABLE_JUPYTER_PROGRESS = os.getenv("SEMANTICA_DISABLE_JUPYTER_PROGRESS", "").s "on", ) + +def _progress_disabled_from_env() -> bool: + """Return whether progress output is disabled for this process.""" + return os.getenv("SEMANTICA_DISABLE_PROGRESS", "").strip().lower() in ( + "1", + "true", + "yes", + "on", + ) + # Try to import IPython for Jupyter support try: from IPython import get_ipython @@ -1013,13 +1023,13 @@ class ProgressTracker: Initialize progress tracker. Args: - enabled: Enable progress tracking (default: True, always enabled) + enabled: Enable progress tracking (default: True) use_emoji: Use emoji indicators update_interval: Minimum time between updates (seconds) """ - # Always enable progress tracking by default - cannot be disabled via constructor - # This ensures progress is always shown automatically - self.enabled = True # Force enabled, ignore parameter + self._progress_forced_disabled = _progress_disabled_from_env() + self._enabled = False + self.enabled = enabled self.use_emoji = use_emoji self.update_interval = update_interval @@ -1061,6 +1071,20 @@ class ProgressTracker: self.pipeline_items: Dict[str, Dict[str, ProgressItem]] = {} # pipeline_id -> {tracking_id: item} self.pipeline_module_order: Dict[str, Dict[str, int]] = {} # pipeline_id -> {module: order} + @property + def enabled(self) -> bool: + """Whether progress tracking is currently enabled.""" + return self._enabled + + @enabled.setter + def enabled(self, value: bool) -> None: + """Set progress enabled unless disabled by environment.""" + if value and (self._progress_forced_disabled or _progress_disabled_from_env()): + self._progress_forced_disabled = True + self._enabled = False + return + self._enabled = bool(value) + def _detect_jupyter(self) -> bool: """Detect if running in Jupyter notebook or Google Colab.""" if not IPYTHON_AVAILABLE: @@ -1114,13 +1138,42 @@ class ProgressTracker: with cls._lock: if cls._instance is None: cls._instance = cls() - # Ensure it's always enabled - cls._instance.enabled = True - else: - # Always ensure enabled when getting instance - cls._instance.enabled = True return cls._instance + def _ensure_jupyter_display(self) -> None: + """Add a Jupyter display dynamically when the environment becomes available.""" + if not self.enabled or not IPYTHON_AVAILABLE or self.is_jupyter: + return + + is_jupyter = self._detect_jupyter() + if not is_jupyter: + return + + with self.lock: + self.is_jupyter = True + if ( + not self.disable_jupyter_progress + and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays) + ): + self.displays.insert(0, JupyterProgressDisplay(use_emoji=self.use_emoji)) + + def _update_displays( + self, + displays: List[ProgressDisplay], + item: ProgressItem, + *, + force_console: bool = False, + ) -> None: + """Update displays without holding the tracker lock.""" + for display in displays: + if force_console and isinstance(display, ConsoleProgressDisplay): + original_last_update = display.last_update + display.last_update = 0.0 + display.update(item) + display.last_update = original_last_update + else: + display.update(item) + def register_pipeline_modules( self, pipeline_id: str, module_list: List[str], module_order: Optional[Dict[str, int]] = None ) -> None: @@ -1217,18 +1270,7 @@ class ProgressTracker: if not self.enabled: return "" - # Re-detect Jupyter environment in case it wasn't detected at init - # This helps if the tracker was created before Jupyter was fully initialized - if IPYTHON_AVAILABLE and not self.is_jupyter: - self.is_jupyter = self._detect_jupyter() - # If Jupyter is now detected and we don't have a Jupyter display, add it - if ( - self.is_jupyter - and not self.disable_jupyter_progress - and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays) - ): - # Insert Jupyter display at the beginning for priority - self.displays.insert(0, JupyterProgressDisplay(use_emoji=self.use_emoji)) + self._ensure_jupyter_display() # Auto-detect if not provided if not module or not submodule: @@ -1241,18 +1283,18 @@ class ProgressTracker: # Create tracking ID tracking_id = f"{module}:{submodule}:{file or ''}" - # Determine pipeline_id and pipeline_order if module is part of a pipeline - pipeline_order = None - if pipeline_id is None and module: - # Try to find pipeline_id from existing contexts - for pid, modules in self.pipeline_contexts.items(): - if module in modules: - pipeline_id = pid - if pid in self.pipeline_module_order: - pipeline_order = self.pipeline_module_order[pid].get(module) - break - with self.lock: + # Determine pipeline_id and pipeline_order if module is part of a pipeline + pipeline_order = None + if pipeline_id is None and module: + # Try to find pipeline_id from existing contexts + for pid, modules in self.pipeline_contexts.items(): + if module in modules: + pipeline_id = pid + if pid in self.pipeline_module_order: + pipeline_order = self.pipeline_module_order[pid].get(module) + break + item = ProgressItem( file=file, module=module, @@ -1273,10 +1315,9 @@ class ProgressTracker: self.pipeline_items[pipeline_id] = {} self.pipeline_items[pipeline_id][tracking_id] = item - # Update displays - for display in self.displays: - display.update(item) + displays = list(self.displays) + self._update_displays(displays, item) return tracking_id def update_tracking( @@ -1293,6 +1334,8 @@ class ProgressTracker: if not self.enabled or not tracking_id: return + item = None + displays: List[ProgressDisplay] = [] with self.lock: if tracking_id in self.active_items: item = self.active_items[tracking_id] @@ -1327,9 +1370,10 @@ class ProgressTracker: self.items.append(item) del self.active_items[tracking_id] - # Update displays - for display in self.displays: - display.update(item) + displays = list(self.displays) + + if item is not None: + self._update_displays(displays, item) def update_progress( self, @@ -1350,18 +1394,10 @@ class ProgressTracker: if not self.enabled or not tracking_id: return - # Re-detect Jupyter environment in case it wasn't detected at init - if IPYTHON_AVAILABLE and not self.is_jupyter: - self.is_jupyter = self._detect_jupyter() - # If Jupyter is now detected and we don't have a Jupyter display, add it - if ( - self.is_jupyter - and not self.disable_jupyter_progress - and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays) - ): - # Insert Jupyter display at the beginning for priority - self.displays.insert(0, JupyterProgressDisplay(use_emoji=self.use_emoji)) + self._ensure_jupyter_display() + item = None + displays: List[ProgressDisplay] = [] with self.lock: if tracking_id in self.active_items: item = self.active_items[tracking_id] @@ -1374,21 +1410,10 @@ class ProgressTracker: if message: item.message = message - # Update displays - force immediate update for progress - for display in self.displays: - # For Jupyter, always update immediately - if isinstance(display, JupyterProgressDisplay): - display.update(item) - # For console, force update by temporarily bypassing interval check - elif isinstance(display, ConsoleProgressDisplay): - # Force update by setting last_update far in the past - original_last_update = display.last_update - display.last_update = 0.0 # This will make _should_update return True - display.update(item) - # Restore original value (update() will set it to current time anyway) - display.last_update = original_last_update - else: - display.update(item) + displays = list(self.displays) + + if item is not None: + self._update_displays(displays, item, force_console=True) def _calculate_eta(self, item: ProgressItem) -> Optional[float]: """ @@ -1495,10 +1520,11 @@ class ProgressTracker: with self.lock: # Add any remaining active items all_items = self.items + list(self.active_items.values()) + displays = list(self.displays) - # Show summary on all displays - for display in self.displays: - display.show_summary(all_items) + # Show summary on all displays without holding the tracker lock. + for display in displays: + display.show_summary(all_items) @contextmanager def track( @@ -1535,21 +1561,8 @@ def get_progress_tracker() -> ProgressTracker: global _global_tracker if _global_tracker is None: _global_tracker = ProgressTracker.get_instance() - - # Always ensure progress tracker is enabled automatically - _global_tracker.enabled = True - - # Re-detect Jupyter environment dynamically (in case it wasn't ready at init) - if IPYTHON_AVAILABLE and not _global_tracker.is_jupyter: - _global_tracker.is_jupyter = _global_tracker._detect_jupyter() - # If Jupyter is now detected and we don't have a Jupyter display, add it - if ( - _global_tracker.is_jupyter - and not _global_tracker.disable_jupyter_progress - and not any(isinstance(d, JupyterProgressDisplay) for d in _global_tracker.displays) - ): - # Insert Jupyter display at the beginning for priority - _global_tracker.displays.insert(0, JupyterProgressDisplay(use_emoji=_global_tracker.use_emoji)) + + _global_tracker._ensure_jupyter_display() return _global_tracker diff --git a/tests/test_cli_foundation.py b/tests/test_cli_foundation.py index e22992cb..92bdae10 100644 --- a/tests/test_cli_foundation.py +++ b/tests/test_cli_foundation.py @@ -1,3 +1,7 @@ +import json +import os +from pathlib import Path + import click import pytest from click.testing import CliRunner, Result @@ -28,13 +32,11 @@ def test_root_help_shows_expected_groups(runner): result = runner.invoke(cli_module.main, ["--help"]) assert result.exit_code == 0 - assert ( - "Semantica - Semantic Layer & Knowledge Engineering Framework" - in result.output - ) + assert "semantica v" in result.output + assert "Knowledge Intelligence Platform" in result.output assert "kg" in result.output assert "pipeline" in result.output - assert "services" in result.output + assert "Services" in result.output def test_kg_group_help_shows_build_command(runner): @@ -279,6 +281,106 @@ def test_build_result_none_shows_generic_success(runner, monkeypatch): assert "Knowledge base build completed" in result.output +@pytest.mark.parametrize( + "argv", + [ + ["--dry-run", "kg", "build", "-s", "src.txt"], + ["--dry-run", "build", "-s", "src.txt"], + ], +) +def test_build_global_dry_run_does_not_initialize_framework(runner, monkeypatch, argv): + """Global --dry-run previews build without touching the framework.""" + + def fail_get_framework(_): + raise AssertionError("_get_framework should not be called during dry-run") + + monkeypatch.setattr(cli_module, "_get_framework", fail_get_framework) + + result = runner.invoke(cli_module.main, argv) + + assert result.exit_code == 0 + assert "Dry run" in result.output + assert "build knowledge base" in result.output + assert "src.txt" in result.output + + +@pytest.mark.parametrize( + "argv", + [ + ["--json", "--dry-run", "kg", "build", "-s", "src.txt"], + ["--json", "--dry-run", "build", "-s", "src.txt"], + ], +) +def test_build_global_dry_run_json(runner, monkeypatch, argv): + """Global --dry-run build emits machine-readable preview JSON.""" + + def fail_get_framework(_): + raise AssertionError("_get_framework should not be called during dry-run") + + monkeypatch.setattr(cli_module, "_get_framework", fail_get_framework) + + result = runner.invoke(cli_module.main, argv) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload == { + "dry_run": True, + "action": "build knowledge base", + "sources": ["src.txt"], + } + + +def _cli_context_with_config(config): + return cli_module.CLIContext( + config_path=None, + config=config, + log_level="INFO", + ) + + +def _runtime_dir(name: str) -> Path: + root = Path(__file__).resolve().parents[1] + work_dir = root / "test_data" / "runtime" / f"{name}-{os.getpid()}" + work_dir.mkdir(parents=True, exist_ok=True) + return work_dir + + +def test_log_directory_probe_does_not_delete_active_log_file(monkeypatch): + """doctor probes with a temp file instead of touching semantica.log.""" + work_dir = _runtime_dir("doctor-log-probe") + monkeypatch.chdir(work_dir) + log_path = work_dir / "semantica.log" + log_path.write_text("existing log", encoding="utf-8") + + config = cli_module._build_runtime_config(None, None) + note = cli_module._check_log_directory(_cli_context_with_config(config)) + + assert "writable" in note + assert log_path.read_text(encoding="utf-8") == "existing log" + + +def test_log_directory_probe_accepts_console_only_logging(): + config = cli_module._build_runtime_config(None, None) + config.set("logging.file", None) + + note = cli_module._check_log_directory(_cli_context_with_config(config)) + + assert note == "console-only logging" + + +def test_doctor_json_log_directory_check_is_not_false_failure(runner, monkeypatch): + work_dir = _runtime_dir("doctor-json") + monkeypatch.chdir(work_dir) + (work_dir / "semantica.log").write_text("existing log", encoding="utf-8") + + result = runner.invoke(cli_module.main, ["--json", "doctor"]) + + assert result.exit_code == 0 + checks = json.loads(result.output) + log_check = next(item for item in checks if item["check"] == "Log directory") + assert log_check["status"] == "ok" + + # --------------------------------------------------------------------------- # Error handling # --------------------------------------------------------------------------- @@ -394,7 +496,8 @@ def test_runtime_errors_are_click_safe_without_traceback(runner, monkeypatch): result = runner.invoke(cli_module.main, ["kg", "build", "-s", "README.md"]) assert result.exit_code != 0 - assert "Unexpected error: boom" in result.output + assert "RuntimeError" in result.output + assert "boom" in result.output assert "Traceback" not in result.output diff --git a/tests/test_progress_tracker_regressions.py b/tests/test_progress_tracker_regressions.py new file mode 100644 index 00000000..9ae11dee --- /dev/null +++ b/tests/test_progress_tracker_regressions.py @@ -0,0 +1,130 @@ +import os +import subprocess +import sys +import threading +from pathlib import Path + +import pytest + +import semantica.utils.progress_tracker as progress_module +from semantica.utils.progress_tracker import ConsoleProgressDisplay, ProgressTracker + + +@pytest.fixture(autouse=True) +def reset_progress_singletons(monkeypatch): + monkeypatch.delenv("SEMANTICA_DISABLE_PROGRESS", raising=False) + ProgressTracker._instance = None + progress_module._global_tracker = None + yield + ProgressTracker._instance = None + progress_module._global_tracker = None + + +def _install_tracker_as_singleton(tracker: ProgressTracker) -> None: + ProgressTracker._instance = tracker + progress_module._global_tracker = tracker + + +def _assert_finishes_quickly(target, timeout: float = 1.0) -> None: + errors = [] + + def runner(): + try: + target() + except BaseException as exc: # pragma: no cover - re-raised below + errors.append(exc) + + thread = threading.Thread(target=runner, daemon=True) + thread.start() + thread.join(timeout) + + assert not thread.is_alive(), "operation deadlocked" + if errors: + raise errors[0] + + +def test_start_tracking_pipeline_console_callback_does_not_deadlock(): + tracker = ProgressTracker(enabled=True, use_emoji=False, update_interval=0) + tracker.displays = [ConsoleProgressDisplay(use_emoji=False, update_interval=0)] + _install_tracker_as_singleton(tracker) + tracker.register_pipeline_modules("pipeline-1", ["core"], {"core": 0}) + + _assert_finishes_quickly( + lambda: tracker.start_tracking( + module="core", + submodule="Semantica", + message="Building", + pipeline_id="pipeline-1", + ) + ) + + +def test_update_progress_pipeline_console_callback_does_not_deadlock(): + tracker = ProgressTracker(enabled=True, use_emoji=False, update_interval=0) + _install_tracker_as_singleton(tracker) + tracker.displays = [] + tracker.register_pipeline_modules("pipeline-1", ["core"], {"core": 0}) + tracking_id = tracker.start_tracking( + module="core", + submodule="Semantica", + message="Building", + pipeline_id="pipeline-1", + ) + tracker.displays = [ConsoleProgressDisplay(use_emoji=False, update_interval=0)] + + _assert_finishes_quickly( + lambda: tracker.update_progress( + tracking_id, + processed=1, + total=1, + message="Processing sources... 1/1", + ) + ) + + +def test_progress_tracker_constructor_can_disable_progress(): + tracker = ProgressTracker(enabled=False) + + assert tracker.enabled is False + assert tracker.start_tracking(module="core", submodule="test") == "" + + +def test_disable_progress_env_prevents_reenable(monkeypatch): + monkeypatch.setenv("SEMANTICA_DISABLE_PROGRESS", "1") + ProgressTracker._instance = None + progress_module._global_tracker = None + + tracker = progress_module.get_progress_tracker() + tracker.enabled = True + + assert tracker.enabled is False + assert tracker.start_tracking(module="core", submodule="test") == "" + + +def test_build_knowledge_base_subprocess_does_not_deadlock(): + root = Path(__file__).resolve().parents[1] + runtime_dir = root / "test_data" / "runtime" / f"build-regression-{os.getpid()}" + runtime_dir.mkdir(parents=True, exist_ok=True) + sample = runtime_dir / "sample.txt" + sample.write_text("Semantica builds small local knowledge graphs.", encoding="utf-8") + + code = ( + "from semantica.core import Semantica; " + f"result = Semantica().build_knowledge_base([{str(sample)!r}], embeddings=False, graph=False); " + "print(result['statistics']['sources_processed'])" + ) + env = os.environ.copy() + env["PYTHONPATH"] = str(root) + result = subprocess.run( + [sys.executable, "-c", code], + cwd=root, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=15, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert "1" in result.stdout