CoreRasurae
cd808dc3dd
/ test (push) Successful in 9m10s
Extend CleverSwarm codebase with PDF and Machine and Worker IDs ISSUES CLOSED: #1
592 lines
25 KiB
Python
592 lines
25 KiB
Python
"""
|
|
Unit tests for cswarm_benchmark_client module.
|
|
|
|
Copyright (c) 2016 - present Syncleus, Inc.
|
|
Copyright (c) 2016 - present CleverThis, Inc.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
"""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import pytest
|
|
|
|
from cleverswarm_python_client.cswarm_benchmark_client import BenchmarkCLI, main
|
|
from cleverswarm_python_client.libs.exceptions import ClientException, JobNotFoundException
|
|
from cleverswarm_python_client.libs.job_enums import JobStatus
|
|
|
|
|
|
class TestBenchmarkCLIInit:
|
|
"""Test cases for BenchmarkCLI initialization."""
|
|
|
|
def test_init_with_defaults(self):
|
|
"""Test initialization with default parameters."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = BenchmarkCLI(temp_dir)
|
|
|
|
assert cli._base_url == "http://localhost:8000/api/v0/"
|
|
assert cli._base_path == Path(temp_dir).resolve()
|
|
assert cli._detailed_metrics is False
|
|
assert cli._client is not None
|
|
|
|
def test_init_with_custom_params(self):
|
|
"""Test initialization with custom parameters."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = BenchmarkCLI(
|
|
temp_dir,
|
|
base_url="http://test:8080/api/",
|
|
detailed_metrics=True,
|
|
username="testuser",
|
|
token="testtoken"
|
|
)
|
|
|
|
assert cli._base_url == "http://test:8080/api/"
|
|
assert cli._base_path == Path(temp_dir).resolve()
|
|
assert cli._detailed_metrics is True
|
|
assert cli._client is not None
|
|
|
|
def test_init_ontologies_mapping(self):
|
|
"""Test that ontologies mapping is correctly initialized."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = BenchmarkCLI(temp_dir)
|
|
|
|
assert len(cli._all_ontologies) == 21
|
|
assert cli._all_ontologies[1] == "1_university"
|
|
assert cli._all_ontologies[21] == "21_complex"
|
|
|
|
def test_init_path_templates(self):
|
|
"""Test that path templates are correctly initialized."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = BenchmarkCLI(temp_dir)
|
|
|
|
assert cli._benchmark_path == "benchmark_data"
|
|
assert cli._input_template == "test_new/ont_$$onto$$_test.jsonl"
|
|
assert cli._onto_template == "ontologies_enriched/$$onto$$_ontology_with_descriptions.json"
|
|
assert cli._ground_truth_template == "ground_truth_new/ont_$$onto$$_ground_truth.jsonl"
|
|
assert cli._output_template == "benchmark_data/outputs_new/$$onto$$_output.jsonl"
|
|
|
|
|
|
class TestBenchmarkCLIOntologyValidation:
|
|
"""Test cases for BenchmarkCLI ontology validation methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = BenchmarkCLI(temp_dir)
|
|
|
|
def test_find_invalid_ontologies_empty_list(self):
|
|
"""Test finding invalid ontologies with empty list."""
|
|
result = self.cli.find_invalid_ontologies([])
|
|
assert result == []
|
|
|
|
def test_find_invalid_ontologies_all_valid(self):
|
|
"""Test finding invalid ontologies with all valid IDs."""
|
|
result = self.cli.find_invalid_ontologies([1, 5, 10, 21])
|
|
assert result == []
|
|
|
|
def test_find_invalid_ontologies_some_invalid(self):
|
|
"""Test finding invalid ontologies with some invalid IDs."""
|
|
result = self.cli.find_invalid_ontologies([1, 5, 99, 10, 100])
|
|
assert result == [99, 100]
|
|
|
|
def test_find_invalid_ontologies_all_invalid(self):
|
|
"""Test finding invalid ontologies with all invalid IDs."""
|
|
result = self.cli.find_invalid_ontologies([99, 100, 101])
|
|
assert result == [99, 100, 101]
|
|
|
|
def test_find_invalid_ontologies_boundary_values(self):
|
|
"""Test finding invalid ontologies with boundary values."""
|
|
# Test valid boundary values
|
|
result = self.cli.find_invalid_ontologies([1, 21])
|
|
assert result == []
|
|
|
|
# Test invalid boundary values
|
|
result = self.cli.find_invalid_ontologies([0, 22])
|
|
assert result == [0, 22]
|
|
|
|
|
|
class TestBenchmarkCLIJobCreation:
|
|
"""Test cases for BenchmarkCLI job creation methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = BenchmarkCLI(temp_dir)
|
|
|
|
def test_create_benchmark_empty_ontologies(self):
|
|
"""Test creating benchmark with empty ontologies list."""
|
|
result = self.cli.create_benchmark([])
|
|
assert result is None
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_create_benchmark_single_ontology(self, mock_client_class):
|
|
"""Test creating benchmark with single ontology."""
|
|
mock_client = Mock()
|
|
mock_client.create_benchmark_job.return_value = "test_job_123"
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create test files
|
|
test_file = Path(temp_dir) / "test_new" / "ont_1_university_test.jsonl"
|
|
test_file.parent.mkdir(parents=True)
|
|
test_file.write_text('{"test": "data"}')
|
|
|
|
onto_file = Path(temp_dir) / "ontologies_enriched" / "1_university_ontology_with_descriptions.json"
|
|
onto_file.parent.mkdir(parents=True)
|
|
onto_file.write_text('{"ontology": "data"}')
|
|
|
|
gt_file = Path(temp_dir) / "ground_truth_new" / "ont_1_university_ground_truth.jsonl"
|
|
gt_file.parent.mkdir(parents=True)
|
|
gt_file.write_text('{"ground_truth": "data"}')
|
|
|
|
# Update base path
|
|
self.cli._base_path = Path(temp_dir)
|
|
|
|
result = self.cli.create_benchmark([1])
|
|
assert result == "test_job_123"
|
|
mock_client.create_benchmark_job.assert_called_once()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_create_benchmark_multiple_ontologies(self, mock_client_class):
|
|
"""Test creating benchmark with multiple ontologies."""
|
|
mock_client = Mock()
|
|
mock_client.create_benchmark_job.return_value = "test_job_123"
|
|
mock_client.append_to_benchmark_job.return_value = True
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create test files for multiple ontologies
|
|
for onto_id in [1, 2]:
|
|
onto_name = self.cli._all_ontologies[onto_id]
|
|
|
|
test_file = Path(temp_dir) / "input_data" / "test_new" / f"ont_{onto_name}_test.jsonl"
|
|
test_file.parent.mkdir(parents=True, exist_ok=True)
|
|
test_file.write_text('{"test": "data"}')
|
|
|
|
onto_file = Path(temp_dir) / "input_data" / "ontologies_enriched" / f"{onto_name}_ontology_with_descriptions.json"
|
|
onto_file.parent.mkdir(parents=True, exist_ok=True)
|
|
onto_file.write_text('{"ontology": "data"}')
|
|
|
|
gt_file = Path(temp_dir) / "input_data" / "ground_truth_new" / f"ont_{onto_name}_ground_truth.jsonl"
|
|
gt_file.parent.mkdir(parents=True, exist_ok=True)
|
|
gt_file.write_text('{"ground_truth": "data"}')
|
|
|
|
# Update base path
|
|
self.cli._base_path = Path(temp_dir)
|
|
|
|
result = self.cli.create_benchmark([1, 2])
|
|
assert result == "test_job_123"
|
|
assert mock_client.create_benchmark_job.call_count == 1
|
|
assert mock_client.append_to_benchmark_job.call_count == 1
|
|
|
|
|
|
class TestBenchmarkCLIJobPolling:
|
|
"""Test cases for BenchmarkCLI job polling methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = BenchmarkCLI(temp_dir)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_poll_job_to_completion_success(self, mock_client_class):
|
|
"""Test successful job polling."""
|
|
mock_client = Mock()
|
|
mock_client.get_job_status.return_value = JobStatus.Completed
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
# Should not raise any exception
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_poll_job_to_completion_failed(self, mock_client_class):
|
|
"""Test job polling with failed job."""
|
|
mock_client = Mock()
|
|
mock_client.get_job_status.return_value = JobStatus.Failed
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with pytest.raises(ClientException, match="finished without completing"):
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_poll_job_to_completion_processing(self, mock_client_class):
|
|
"""Test job polling with processing job."""
|
|
mock_client = Mock()
|
|
mock_client.get_job_status.return_value = JobStatus.Processing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Completed
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
# Should not raise any exception
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_poll_job_to_completion_ready_for_processing(self, mock_client_class):
|
|
"""Test job polling with ready for processing job."""
|
|
mock_client = Mock()
|
|
mock_client.get_job_status.return_value = JobStatus.ReadyForProcessing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Completed
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
# Should not raise any exception
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_poll_job_to_completion_invalid_status(self, mock_client_class):
|
|
"""Test job polling with invalid status."""
|
|
mock_client = Mock()
|
|
mock_client.get_job_status.return_value = JobStatus.Created
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with pytest.raises(ClientException, match="is not processing"):
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_poll_job_to_completion_poll_failed(self, mock_client_class):
|
|
"""Test job polling where poll returns failed."""
|
|
mock_client = Mock()
|
|
mock_client.get_job_status.return_value = JobStatus.Processing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Failed
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with pytest.raises(ClientException, match="finished without completing"):
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
|
|
class TestBenchmarkCLIJobDownload:
|
|
"""Test cases for BenchmarkCLI job download methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = BenchmarkCLI(temp_dir)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_download_job_results_success(self, mock_client_class):
|
|
"""Test successful job results download."""
|
|
mock_client = Mock()
|
|
mock_client.retrieve_benchmark_files.return_value = True
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli._base_path = Path(temp_dir)
|
|
|
|
# Should not raise any exception
|
|
self.cli.download_job_results("test_job_123", [1, 2])
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_download_job_results_creates_output_dirs(self, mock_client_class):
|
|
"""Test that download creates necessary output directories."""
|
|
mock_client = Mock()
|
|
mock_client.retrieve_benchmark_files.return_value = True
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli._base_path = Path(temp_dir)
|
|
|
|
self.cli.download_job_results("test_job_123", [1])
|
|
|
|
# Check that retrieve_benchmark_files was called with correct paths
|
|
mock_client.retrieve_benchmark_files.assert_called_once()
|
|
call_args = mock_client.retrieve_benchmark_files.call_args
|
|
assert call_args[0][0] == "test_job_123" # job_id
|
|
output_files = call_args[0][1] # output_filenames
|
|
assert len(output_files) == 1
|
|
assert "university_output.jsonl" in str(output_files[0])
|
|
|
|
|
|
class TestBenchmarkCLIJobManagement:
|
|
"""Test cases for BenchmarkCLI job management methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = BenchmarkCLI(temp_dir)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_list_server_jobs_success(self, mock_client_class):
|
|
"""Test successful server jobs listing."""
|
|
mock_client = Mock()
|
|
mock_jobs = [
|
|
{"id": "job1", "status": "Completed"},
|
|
{"id": "job2", "status": "Processing"}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
result = self.cli.list_server_jobs()
|
|
assert result == mock_jobs
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_extract_server_job_ontologies_success(self, mock_client_class):
|
|
"""Test successful extraction of server job ontologies."""
|
|
mock_client = Mock()
|
|
mock_job_details = {
|
|
"ontology_sources": ["1_university_ontology.json", "2_musicalwork_ontology.json"]
|
|
}
|
|
mock_client.get_job_details.return_value = mock_job_details
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
result = self.cli.extract_server_job_ontologies("test_job_123")
|
|
assert result == [1, 2]
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_extract_server_job_ontologies_not_found(self, mock_client_class):
|
|
"""Test extraction of ontologies from non-existent job."""
|
|
mock_client = Mock()
|
|
mock_client.get_job_details.return_value = None
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
with pytest.raises(JobNotFoundException):
|
|
self.cli.extract_server_job_ontologies("nonexistent_job")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_delete_server_job_success(self, mock_client_class):
|
|
"""Test successful server job deletion."""
|
|
mock_client = Mock()
|
|
mock_client.delete_job.return_value = True
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
# Should not raise any exception
|
|
self.cli.delete_server_job("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_delete_server_job_failure(self, mock_client_class):
|
|
"""Test server job deletion failure."""
|
|
mock_client = Mock()
|
|
mock_client.delete_job.return_value = False
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
# Should not raise any exception, but should handle failure gracefully
|
|
self.cli.delete_server_job("test_job_123")
|
|
|
|
|
|
class TestBenchmarkCLIEvaluation:
|
|
"""Test cases for BenchmarkCLI evaluation methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = BenchmarkCLI(temp_dir)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.EvaluateResults')
|
|
def test_evaluate_results_locally_success(self, mock_evaluate_results):
|
|
"""Test successful local results evaluation."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli._base_path = Path(temp_dir)
|
|
|
|
result = self.cli.evaluate_results_locally([1, 2])
|
|
|
|
assert result == self.cli._base_path / "benchmark_data" / "local_metrics"
|
|
mock_evaluate_results.assert_called_once()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.EvaluateResults')
|
|
def test_evaluate_results_locally_detailed_metrics(self, mock_evaluate_results):
|
|
"""Test local results evaluation with detailed metrics."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli._base_path = Path(temp_dir)
|
|
self.cli._detailed_metrics = True
|
|
|
|
result = self.cli.evaluate_results_locally([1, 2])
|
|
|
|
assert result == self.cli._base_path / "benchmark_data" / "local_metrics"
|
|
# Check that detailed metrics flag is passed
|
|
call_args = mock_evaluate_results.call_args
|
|
assert call_args[0][1] is True # is_detailed_metrics parameter
|
|
|
|
|
|
class TestBenchmarkCLIUtilityMethods:
|
|
"""Test cases for BenchmarkCLI utility methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = BenchmarkCLI(temp_dir)
|
|
|
|
def test_print_ontologies_ids(self, capsys):
|
|
"""Test printing ontology IDs."""
|
|
self.cli.print_ontologies_ids()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "ID, Name" in captured.out
|
|
assert "1, 1_university" in captured.out
|
|
assert "21, 21_complex" in captured.out
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_print_server_jobs_basic(self, mock_client_class, capsys):
|
|
"""Test printing server jobs in basic mode."""
|
|
mock_client = Mock()
|
|
mock_jobs = [
|
|
{
|
|
"id": "job1",
|
|
"created": "2023-01-01T00:00:00Z",
|
|
"type": "Benchmark",
|
|
"status": "Completed",
|
|
"retries_count": 0,
|
|
"file_type": "JSONL",
|
|
"machine_id": "machine-001",
|
|
"worker_id": "worker-001"
|
|
}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
self.cli.print_server_jobs()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Job ID: job1" in captured.out
|
|
assert "Status: Completed" in captured.out
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.CleverSwarmClient')
|
|
def test_print_server_jobs_detailed(self, mock_client_class, capsys):
|
|
"""Test printing server jobs in detailed mode."""
|
|
mock_client = Mock()
|
|
mock_jobs = [
|
|
{
|
|
"id": "job1",
|
|
"created": "2023-01-01T00:00:00Z",
|
|
"type": "Benchmark",
|
|
"status": "Completed",
|
|
"retries_count": 0,
|
|
"file_type": "JSONL",
|
|
"machine_id": "machine-001",
|
|
"worker_id": "worker-001",
|
|
"unstructured_sources": ["test.txt"],
|
|
"ontology_sources": ["test.json"],
|
|
"ground_truth_sources": ["gt.jsonl"]
|
|
}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
self.cli.print_server_jobs(detailed=True)
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Job ID: job1" in captured.out
|
|
assert "Unstructured text input files:" in captured.out
|
|
assert "Ontology input files:" in captured.out
|
|
assert "Ground-truth input files:" in captured.out
|
|
|
|
|
|
class TestBenchmarkCLIMainFunction:
|
|
"""Test cases for BenchmarkCLI main function."""
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.BenchmarkCLI')
|
|
def test_main_list_ontologies(self, mock_cli_class):
|
|
"""Test main function with list ontologies action."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_benchmark_client.py', '--action', 'list-ontologies-ids-and-exit']):
|
|
main()
|
|
|
|
mock_cli.print_ontologies_ids.assert_called_once()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.BenchmarkCLI')
|
|
def test_main_create_and_exit(self, mock_cli_class):
|
|
"""Test main function with create and exit action."""
|
|
mock_cli = Mock()
|
|
mock_cli.create_benchmark.return_value = "test_job_123"
|
|
mock_cli.find_invalid_ontologies.return_value = []
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_benchmark_client.py', '--action', 'create-and-exit', '--ontologies_ids', '1', '2']):
|
|
main() # Should not raise SystemExit on success
|
|
|
|
mock_cli.create_benchmark.assert_called_once_with([1, 2])
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.BenchmarkCLI')
|
|
def test_main_invalid_ontologies(self, mock_cli_class):
|
|
"""Test main function with invalid ontology IDs."""
|
|
mock_cli = Mock()
|
|
mock_cli.find_invalid_ontologies.return_value = [99, 100]
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_benchmark_client.py', '--action', 'create-and-exit', '--ontologies_ids', '1', '99', '100']):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.BenchmarkCLI')
|
|
def test_main_client_exception(self, mock_cli_class):
|
|
"""Test main function with client exception."""
|
|
mock_cli = Mock()
|
|
mock_cli.find_invalid_ontologies.side_effect = ClientException("Test error")
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_benchmark_client.py', '--action', 'create-and-exit', '--ontologies_ids', '1']):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_benchmark_client.BenchmarkCLI')
|
|
def test_main_unexpected_exception(self, mock_cli_class):
|
|
"""Test main function with unexpected exception."""
|
|
mock_cli = Mock()
|
|
mock_cli.find_invalid_ontologies.side_effect = Exception("Unexpected error")
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_benchmark_client.py', '--action', 'create-and-exit', '--ontologies_ids', '1']):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
|
|
class TestBenchmarkCLIIntegration:
|
|
"""Integration test cases for BenchmarkCLI."""
|
|
|
|
def test_full_workflow_simulation(self):
|
|
"""Test a complete workflow simulation."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = BenchmarkCLI(temp_dir)
|
|
|
|
# Test ontology validation
|
|
invalid_ids = cli.find_invalid_ontologies([1, 2, 99])
|
|
assert invalid_ids == [99]
|
|
|
|
# Test valid ontologies
|
|
valid_ids = cli.find_invalid_ontologies([1, 2, 3])
|
|
assert valid_ids == []
|
|
|
|
# Test printing ontologies
|
|
with patch('builtins.print') as mock_print:
|
|
cli.print_ontologies_ids()
|
|
assert mock_print.call_count > 0
|
|
|
|
def test_path_template_substitution(self):
|
|
"""Test path template substitution."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = BenchmarkCLI(temp_dir)
|
|
|
|
# Test template substitution
|
|
onto_name = cli._all_ontologies[1]
|
|
input_path = cli._input_template.replace("$$onto$$", onto_name)
|
|
assert input_path == f"test_new/ont_{onto_name}_test.jsonl"
|
|
|
|
onto_path = cli._onto_template.replace("$$onto$$", onto_name)
|
|
assert onto_path == f"ontologies_enriched/{onto_name}_ontology_with_descriptions.json"
|