Add comprehensive test coverage for missing modules and core utilities

tests/commands/test_crawlinginfo.py

@@ -0,0 +1,119 @@
+import json
+import pytest
+from typer.testing import CliRunner
+
+from fessctl.commands.crawlinginfo import crawlinginfo_app
+
+
+@pytest.fixture(scope="module")
+def runner():
+    """
+    Provides a CliRunner instance for invoking commands.
+    """
+    return CliRunner()
+
+
+def test_crawlinginfo_list(runner, fess_service):
+    """
+    Test listing crawling info entries.
+    CrawlingInfo is typically populated by crawler runs, so we just verify
+    that the list command executes successfully.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["list", "--output", "json"]
+    )
+    assert result.exit_code == 0, f"List failed: {result.stdout}"
+    list_resp = json.loads(result.stdout)
+    assert list_resp.get("response", {}).get("status") == 0
+    # logs may be empty if no crawls have run
+    logs = list_resp["response"].get("logs", [])
+    assert isinstance(logs, list)
+
+
+def test_crawlinginfo_list_with_pagination(runner, fess_service):
+    """
+    Test listing crawling info entries with pagination options.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["list", "--page", "1", "--size", "10", "--output", "json"]
+    )
+    assert result.exit_code == 0, f"List with pagination failed: {result.stdout}"
+    list_resp = json.loads(result.stdout)
+    assert list_resp.get("response", {}).get("status") == 0
+
+
+def test_crawlinginfo_list_yaml_output(runner, fess_service):
+    """
+    Test listing crawling info entries with YAML output format.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["list", "--output", "yaml"]
+    )
+    assert result.exit_code == 0, f"List YAML failed: {result.stdout}"
+    # YAML output should contain 'response:' key
+    assert "response:" in result.stdout or "logs:" in result.stdout
+
+
+def test_crawlinginfo_list_text_output(runner, fess_service):
+    """
+    Test listing crawling info entries with text output format.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["list", "--output", "text"]
+    )
+    # Should succeed even with text output
+    assert result.exit_code == 0, f"List text failed: {result.stdout}"
+
+
+def test_crawlinginfo_get_nonexistent(runner, fess_service):
+    """
+    Test getting a non-existent crawling info entry returns an error.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["get", "nonexistent-id-12345"]
+    )
+    assert result.exit_code != 0
+    assert "failed to retrieve crawlinginfo" in result.stdout.lower()
+
+
+def test_crawlinginfo_get_nonexistent_json_output(runner, fess_service):
+    """
+    Test getting a non-existent crawling info entry with JSON output.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["get", "nonexistent-id-12345", "--output", "json"]
+    )
+    # JSON output always returns, but status should indicate failure
+    get_resp = json.loads(result.stdout)
+    assert get_resp.get("response", {}).get("status") != 0
+
+
+def test_crawlinginfo_delete_nonexistent(runner, fess_service):
+    """
+    Test deleting a non-existent crawling info entry returns an error.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["delete", "nonexistent-id-12345"]
+    )
+    assert result.exit_code != 0
+    assert "failed to delete crawlinginfo" in result.stdout.lower()
+
+
+def test_crawlinginfo_delete_nonexistent_json_output(runner, fess_service):
+    """
+    Test deleting a non-existent crawling info entry with JSON output.
+    """
+    result = runner.invoke(
+        crawlinginfo_app,
+        ["delete", "nonexistent-id-12345", "--output", "json"]
+    )
+    # JSON output always returns, but status should indicate failure
+    del_resp = json.loads(result.stdout)
+    assert del_resp.get("response", {}).get("status") != 0

tests/commands/test_joblog.py

@@ -0,0 +1,119 @@
+import json
+import pytest
+from typer.testing import CliRunner
+
+from fessctl.commands.joblog import joblog_app
+
+
+@pytest.fixture(scope="module")
+def runner():
+    """
+    Provides a CliRunner instance for invoking commands.
+    """
+    return CliRunner()
+
+
+def test_joblog_list(runner, fess_service):
+    """
+    Test listing job log entries.
+    JobLogs are populated by scheduler runs, so we just verify
+    that the list command executes successfully.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["list", "--output", "json"]
+    )
+    assert result.exit_code == 0, f"List failed: {result.stdout}"
+    list_resp = json.loads(result.stdout)
+    assert list_resp.get("response", {}).get("status") == 0
+    # logs may be empty if no jobs have run
+    logs = list_resp["response"].get("logs", [])
+    assert isinstance(logs, list)
+
+
+def test_joblog_list_with_pagination(runner, fess_service):
+    """
+    Test listing job log entries with pagination options.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["list", "--page", "1", "--size", "10", "--output", "json"]
+    )
+    assert result.exit_code == 0, f"List with pagination failed: {result.stdout}"
+    list_resp = json.loads(result.stdout)
+    assert list_resp.get("response", {}).get("status") == 0
+
+
+def test_joblog_list_yaml_output(runner, fess_service):
+    """
+    Test listing job log entries with YAML output format.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["list", "--output", "yaml"]
+    )
+    assert result.exit_code == 0, f"List YAML failed: {result.stdout}"
+    # YAML output should contain 'response:' key
+    assert "response:" in result.stdout or "logs:" in result.stdout
+
+
+def test_joblog_list_text_output(runner, fess_service):
+    """
+    Test listing job log entries with text output format.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["list", "--output", "text"]
+    )
+    # Should succeed even with text output
+    assert result.exit_code == 0, f"List text failed: {result.stdout}"
+
+
+def test_joblog_get_nonexistent(runner, fess_service):
+    """
+    Test getting a non-existent job log entry returns an error.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["get", "nonexistent-id-12345"]
+    )
+    assert result.exit_code != 0
+    assert "failed to retrieve joblog" in result.stdout.lower()
+
+
+def test_joblog_get_nonexistent_json_output(runner, fess_service):
+    """
+    Test getting a non-existent job log entry with JSON output.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["get", "nonexistent-id-12345", "--output", "json"]
+    )
+    # JSON output always returns, but status should indicate failure
+    get_resp = json.loads(result.stdout)
+    assert get_resp.get("response", {}).get("status") != 0
+
+
+def test_joblog_delete_nonexistent(runner, fess_service):
+    """
+    Test deleting a non-existent job log entry returns an error.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["delete", "nonexistent-id-12345"]
+    )
+    assert result.exit_code != 0
+    assert "failed to delete joblog" in result.stdout.lower()
+
+
+def test_joblog_delete_nonexistent_json_output(runner, fess_service):
+    """
+    Test deleting a non-existent job log entry with JSON output.
+    """
+    result = runner.invoke(
+        joblog_app,
+        ["delete", "nonexistent-id-12345", "--output", "json"]
+    )
+    # JSON output always returns, but status should indicate failure
+    del_resp = json.loads(result.stdout)
+    assert del_resp.get("response", {}).get("status") != 0

tests/commands/test_scheduler.py

@@ -93,3 +93,138 @@ def test_scheduler_crud_flow(runner, fess_service):
     )
     assert result.exit_code != 0
     assert "failed to retrieve scheduler" in result.stdout.lower()
+
+
+def test_scheduler_start_stop_json_output(runner, fess_service):
+    """
+    Tests the start and stop commands return valid JSON responses.
+    Note: Start/stop may fail for schedulers without valid script data,
+    so we verify the command returns a valid response structure.
+    """
+    # 1) Create a new scheduler for testing start/stop
+    unique_name = f"schedule-startstop-{uuid.uuid4().hex[:8]}"
+    target = "all"
+    script_type = "groovy"
+    cron = "0 0 * * *"  # Daily at midnight
+    result = runner.invoke(
+        scheduler_app,
+        ["create", "--name", unique_name, "--target", target, "--script-type", script_type, "--cron-expression", cron, "--output", "json"]
+    )
+    assert result.exit_code == 0, f"Create failed: {result.stdout}"
+    create_resp = json.loads(result.stdout)
+    assert create_resp.get("response", {}).get("status") == 0
+    scheduler_id = create_resp["response"].get("id")
+    assert scheduler_id, "No scheduler ID returned on create"
+
+    try:
+        # 2) Start the scheduler - verify returns valid JSON with response structure
+        result = runner.invoke(
+            scheduler_app,
+            ["start", scheduler_id, "--output", "json"]
+        )
+        # Command should return JSON regardless of success/failure
+        start_resp = json.loads(result.stdout)
+        assert "response" in start_resp
+        assert "status" in start_resp["response"]
+
+        # 3) Stop the scheduler - verify returns valid JSON with response structure
+        result = runner.invoke(
+            scheduler_app,
+            ["stop", scheduler_id, "--output", "json"]
+        )
+        stop_resp = json.loads(result.stdout)
+        assert "response" in stop_resp
+        assert "status" in stop_resp["response"]
+
+    finally:
+        # 4) Clean up - delete the scheduler
+        runner.invoke(
+            scheduler_app,
+            ["delete", scheduler_id, "--output", "json"]
+        )
+
+
+def test_scheduler_start_text_output(runner, fess_service):
+    """
+    Tests the start command with text output format.
+    Verifies the command produces output (success or failure message).
+    """
+    # Create a scheduler
+    unique_name = f"schedule-text-{uuid.uuid4().hex[:8]}"
+    result = runner.invoke(
+        scheduler_app,
+        ["create", "--name", unique_name, "--target", "all", "--script-type", "groovy", "--cron-expression", "0 0 * * *", "--output", "json"]
+    )
+    assert result.exit_code == 0
+    scheduler_id = json.loads(result.stdout)["response"]["id"]
+
+    try:
+        # Start with text output - should produce some output
+        result = runner.invoke(
+            scheduler_app,
+            ["start", scheduler_id, "--output", "text"]
+        )
+        # Verify there is output (either success or failure message)
+        assert len(result.stdout) > 0
+        assert "scheduler" in result.stdout.lower()
+
+        # Stop with text output - should produce some output
+        result = runner.invoke(
+            scheduler_app,
+            ["stop", scheduler_id, "--output", "text"]
+        )
+        assert len(result.stdout) > 0
+        assert "scheduler" in result.stdout.lower()
+
+    finally:
+        runner.invoke(scheduler_app, ["delete", scheduler_id])
+
+
+def test_scheduler_start_yaml_output(runner, fess_service):
+    """
+    Tests the start command with YAML output format.
+    """
+    # Create a scheduler
+    unique_name = f"schedule-yaml-{uuid.uuid4().hex[:8]}"
+    result = runner.invoke(
+        scheduler_app,
+        ["create", "--name", unique_name, "--target", "all", "--script-type", "groovy", "--cron-expression", "0 0 * * *", "--output", "json"]
+    )
+    assert result.exit_code == 0
+    scheduler_id = json.loads(result.stdout)["response"]["id"]
+
+    try:
+        # Start with YAML output - verify YAML structure
+        result = runner.invoke(
+            scheduler_app,
+            ["start", scheduler_id, "--output", "yaml"]
+        )
+        # YAML output should contain response key
+        assert "response:" in result.stdout
+
+    finally:
+        runner.invoke(scheduler_app, ["delete", scheduler_id])
+
+
+def test_scheduler_start_nonexistent(runner, fess_service):
+    """
+    Tests starting a non-existent scheduler returns an error.
+    """
+    result = runner.invoke(
+        scheduler_app,
+        ["start", "nonexistent-scheduler-id"]
+    )
+    assert result.exit_code != 0
+    assert "failed to start scheduler" in result.stdout.lower()
+
+
+def test_scheduler_stop_nonexistent(runner, fess_service):
+    """
+    Tests stopping a non-existent scheduler returns an error.
+    """
+    result = runner.invoke(
+        scheduler_app,
+        ["stop", "nonexistent-scheduler-id"]
+    )
+    assert result.exit_code != 0
+    assert "failed to stop scheduler" in result.stdout.lower()