CoreRasurae
cd808dc3dd
/ test (push) Successful in 9m10s
Extend CleverSwarm codebase with PDF and Machine and Worker IDs ISSUES CLOSED: #1
461 lines
20 KiB
Python
461 lines
20 KiB
Python
"""
|
|
Extended unit tests for cswarm_text_to_kg_client module to achieve higher coverage.
|
|
|
|
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
|
|
import pytest
|
|
|
|
from cleverswarm_python_client.cswarm_text_to_kg_client import TextoToKGCLI, main
|
|
from cleverswarm_python_client.libs.exceptions import ClientException
|
|
from cleverswarm_python_client.libs.job_enums import JobStatus
|
|
|
|
|
|
class TestTextoToKGCLIExtended:
|
|
"""Extended test cases for TextoToKGCLI to achieve higher coverage."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test CLI."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
self.cli = TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix=temp_dir,
|
|
output_prefix=temp_dir
|
|
)
|
|
|
|
def test_create_job_with_wildcards_and_force_filetype(self):
|
|
"""Test job creation with wildcards and force filetype."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create test files
|
|
text_file = Path(temp_dir) / "test.txt"
|
|
json_file = Path(temp_dir) / "test.json"
|
|
owl_file = Path(temp_dir) / "test.owl"
|
|
wildcards_file = Path(temp_dir) / "test.wildcards"
|
|
|
|
text_file.write_text("Test content")
|
|
json_file.write_text('{"test": "data"}')
|
|
owl_file.write_text('<rdf:RDF>test</rdf:RDF>')
|
|
wildcards_file.write_text("wildcard1\nwildcard2")
|
|
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.create_unstructured_to_kg_wildcards_job.return_value = "test_job_123"
|
|
|
|
result = self.cli.create_job(
|
|
str(text_file),
|
|
str(json_file),
|
|
str(owl_file),
|
|
wildcards=str(wildcards_file),
|
|
force_filetype="Markdown"
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
mock_client.create_unstructured_to_kg_wildcards_job.assert_called_once()
|
|
|
|
def test_create_job_with_force_filetype_enum(self):
|
|
"""Test job creation with force filetype as enum."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create test files
|
|
text_file = Path(temp_dir) / "test.txt"
|
|
json_file = Path(temp_dir) / "test.json"
|
|
owl_file = Path(temp_dir) / "test.owl"
|
|
|
|
text_file.write_text("Test content")
|
|
json_file.write_text('{"test": "data"}')
|
|
owl_file.write_text('<rdf:RDF>test</rdf:RDF>')
|
|
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.create_unstructured_to_kg_job.return_value = "test_job_123"
|
|
|
|
from cleverswarm_python_client.libs.file_type_enum import FileTypeAPI
|
|
|
|
result = self.cli.create_job(
|
|
str(text_file),
|
|
str(json_file),
|
|
str(owl_file),
|
|
force_filetype=FileTypeAPI.Markdown
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
mock_client.create_unstructured_to_kg_job.assert_called_once()
|
|
|
|
def test_download_job_result_with_single_file(self):
|
|
"""Test downloading job result with single file."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_job_details = {
|
|
"unstructured_sources": ["test.txt"]
|
|
}
|
|
mock_client.get_job_details.return_value = mock_job_details
|
|
mock_client.retrieve_unstructured_to_kg_files.return_value = True
|
|
|
|
result = self.cli.download_job_result("test_job_123")
|
|
|
|
assert result is None
|
|
mock_client.get_job_details.assert_called_once_with("test_job_123")
|
|
mock_client.retrieve_unstructured_to_kg_files.assert_called_once()
|
|
|
|
def test_download_job_result_with_multiple_files_raises_exception(self):
|
|
"""Test downloading job result with multiple files raises exception."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_job_details = {
|
|
"unstructured_sources": ["test1.txt", "test2.txt"]
|
|
}
|
|
mock_client.get_job_details.return_value = mock_job_details
|
|
|
|
with pytest.raises(ClientException, match="has more than one input files"):
|
|
self.cli.download_job_result("test_job_123")
|
|
|
|
def test_download_job_result_with_no_files_raises_exception(self):
|
|
"""Test downloading job result with no files raises exception."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_job_details = {
|
|
"unstructured_sources": []
|
|
}
|
|
mock_client.get_job_details.return_value = mock_job_details
|
|
|
|
with pytest.raises(ClientException, match="has no input files"):
|
|
self.cli.download_job_result("test_job_123")
|
|
|
|
def test_download_job_result_detailed_output(self):
|
|
"""Test downloading job result with detailed output."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
self.cli._detailed = True
|
|
mock_job_details = {
|
|
"unstructured_sources": ["test.txt"]
|
|
}
|
|
mock_client.get_job_details.return_value = mock_job_details
|
|
mock_client.retrieve_unstructured_to_kg_files.return_value = True
|
|
|
|
with patch('builtins.print') as mock_print:
|
|
result = self.cli.download_job_result("test_job_123")
|
|
|
|
assert result is None
|
|
mock_print.assert_called()
|
|
|
|
def test_list_server_jobs_success(self):
|
|
"""Test listing server jobs successfully."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_jobs = [
|
|
{"id": "job1", "type": "UnstructuredWithOntology"},
|
|
{"id": "job2", "type": "UnstructuredWithWildcards"}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
|
|
result = self.cli.list_server_jobs()
|
|
|
|
assert len(result) == 2
|
|
assert result[0]["id"] == "job1"
|
|
mock_client.get_jobs_list.assert_called_once_with(is_benchmark=False)
|
|
|
|
def test_print_server_jobs_with_empty_list(self):
|
|
"""Test printing server jobs with empty list."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.get_jobs_list.return_value = []
|
|
|
|
with patch('builtins.print') as mock_print:
|
|
self.cli.print_server_jobs()
|
|
# Should not raise any exception
|
|
|
|
def test_print_server_jobs_with_jobs(self):
|
|
"""Test printing server jobs with actual jobs."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_jobs = [
|
|
{
|
|
"id": "job1",
|
|
"created": "2023-01-01T00:00:00Z",
|
|
"type": "UnstructuredWithOntology",
|
|
"status": "Completed",
|
|
"retries_count": 0,
|
|
"file_type": "JSONL",
|
|
"machine_id": "machine-001",
|
|
"worker_id": "worker-001",
|
|
"unstructured_sources": ["test.txt"],
|
|
"ontology_sources": ["test.json"],
|
|
"ontology_spec_sources": ["test.owl"]
|
|
}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
|
|
with patch('builtins.print') as mock_print:
|
|
self.cli.print_server_jobs()
|
|
assert mock_print.call_count > 0
|
|
|
|
def test_print_server_jobs_detailed_with_wildcards(self):
|
|
"""Test printing server jobs with detailed output including wildcards."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_jobs = [
|
|
{
|
|
"id": "job1",
|
|
"created": "2023-01-01T00:00:00Z",
|
|
"type": "UnstructuredWithWildcards",
|
|
"status": "Completed",
|
|
"retries_count": 0,
|
|
"file_type": "JSONL",
|
|
"machine_id": "machine-001",
|
|
"worker_id": "worker-001",
|
|
"unstructured_sources": ["test.txt"],
|
|
"ontology_sources": ["test.json"],
|
|
"ontology_spec_sources": ["test.owl"],
|
|
"wildcards_sources": ["test.wildcards"]
|
|
}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
|
|
with patch('builtins.print') as mock_print:
|
|
self.cli.print_server_jobs(is_detailed=True)
|
|
assert mock_print.call_count > 0
|
|
|
|
def test_print_server_jobs_detailed_without_wildcards(self):
|
|
"""Test printing server jobs with detailed output without wildcards."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_jobs = [
|
|
{
|
|
"id": "job1",
|
|
"created": "2023-01-01T00:00:00Z",
|
|
"type": "UnstructuredWithOntology",
|
|
"status": "Completed",
|
|
"retries_count": 0,
|
|
"file_type": "JSONL",
|
|
"machine_id": "machine-001",
|
|
"worker_id": "worker-001",
|
|
"unstructured_sources": ["test.txt"],
|
|
"ontology_sources": ["test.json"],
|
|
"ontology_spec_sources": ["test.owl"]
|
|
}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
|
|
with patch('builtins.print') as mock_print:
|
|
self.cli.print_server_jobs(is_detailed=True)
|
|
assert mock_print.call_count > 0
|
|
|
|
def test_delete_server_job_success(self):
|
|
"""Test successful server job deletion."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.delete_job.return_value = True
|
|
|
|
with patch('builtins.print') as mock_print:
|
|
self.cli.delete_server_job("test_job_123")
|
|
mock_print.assert_called_with("Job was successfully deleted")
|
|
|
|
def test_delete_server_job_failure(self):
|
|
"""Test failed server job deletion."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.delete_job.return_value = False
|
|
|
|
with patch('builtins.print') as mock_print:
|
|
self.cli.delete_server_job("test_job_123")
|
|
mock_print.assert_called_with("Failed to delete Job")
|
|
|
|
def test_poll_job_to_completion_with_ready_for_processing(self):
|
|
"""Test polling job that is ready for processing."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.get_job_status.return_value = JobStatus.ReadyForProcessing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Completed
|
|
|
|
result = self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
assert result is None # Should not raise exception
|
|
mock_client.get_job_status.assert_called_once_with("test_job_123")
|
|
mock_client.poll_job_ready_or_failed.assert_called_once_with("test_job_123")
|
|
|
|
def test_poll_job_to_completion_with_processing(self):
|
|
"""Test polling job that is processing."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.get_job_status.return_value = JobStatus.Processing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Completed
|
|
|
|
result = self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
assert result is None # Should not raise exception
|
|
mock_client.get_job_status.assert_called_once_with("test_job_123")
|
|
mock_client.poll_job_ready_or_failed.assert_called_once_with("test_job_123")
|
|
|
|
def test_poll_job_to_completion_with_invalid_status(self):
|
|
"""Test polling job with invalid status."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.get_job_status.return_value = "InvalidStatus"
|
|
|
|
with pytest.raises(ClientException, match="is not processing, nor is it ready for processing"):
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
def test_poll_job_to_completion_with_poll_failure(self):
|
|
"""Test polling job with poll failure."""
|
|
with patch.object(self.cli, '_client') as mock_client:
|
|
mock_client.get_job_status.return_value = JobStatus.Processing
|
|
mock_client.poll_job_ready_or_failed.side_effect = ClientException("Poll failed")
|
|
|
|
with pytest.raises(ClientException, match="Poll failed"):
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
|
|
class TestTextoToKGCLIMainFunctionExtended:
|
|
"""Extended test cases for TextoToKGCLI main function to achieve higher coverage."""
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_delete_and_exit(self, mock_cli_class):
|
|
"""Test main function with delete and exit action."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'delete-and-exit', '--job_id', 'test_job_123']):
|
|
main()
|
|
|
|
mock_cli.delete_server_job.assert_called_once_with("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_delete_and_exit_missing_job_id(self, mock_cli_class):
|
|
"""Test main function with delete and exit action missing job_id."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'delete-and-exit']):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_download_kg_and_exit(self, mock_cli_class):
|
|
"""Test main function with download-kg-and-exit action."""
|
|
mock_cli = Mock()
|
|
mock_cli.download_job_result.return_value = True
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'download-kg-and-exit', '--job_id', 'test_job_123']):
|
|
main()
|
|
|
|
mock_cli.download_job_result.assert_called_once_with("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_download_kg_and_exit_missing_job_id(self, mock_cli_class):
|
|
"""Test main function with download-kg-and-exit action missing job_id."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'download-kg-and-exit']):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_list_kg_jobs_and_exit(self, mock_cli_class):
|
|
"""Test main function with list-kg-jobs-and-exit action."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'list-kg-jobs-and-exit']):
|
|
main()
|
|
|
|
mock_cli.print_server_jobs.assert_called_once_with(False)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_create_download_kg(self, mock_cli_class):
|
|
"""Test main function with create-download-kg action."""
|
|
mock_cli = Mock()
|
|
mock_cli.create_job.return_value = "test_job_123"
|
|
mock_cli.poll_job_to_completion.return_value = None
|
|
mock_cli.download_job_result.return_value = True
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', [
|
|
'cswarm_text_to_kg_client.py',
|
|
'--action', 'create-download-kg',
|
|
'--unstructured_text', 'test.txt',
|
|
'--ontology_json', 'test.json',
|
|
'--ontology_owl', 'test.owl'
|
|
]):
|
|
main()
|
|
|
|
mock_cli.create_job.assert_called_once()
|
|
mock_cli.poll_job_to_completion.assert_called_once_with("test_job_123")
|
|
mock_cli.download_job_result.assert_called_once_with("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_with_wildcards(self, mock_cli_class):
|
|
"""Test main function with wildcards."""
|
|
mock_cli = Mock()
|
|
mock_cli.create_job.return_value = "test_job_123"
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', [
|
|
'cswarm_text_to_kg_client.py',
|
|
'--action', 'create-and-exit',
|
|
'--unstructured_text', 'test.txt',
|
|
'--ontology_json', 'test.json',
|
|
'--ontology_owl', 'test.owl',
|
|
'--wildcards', 'test.wildcards'
|
|
]):
|
|
main()
|
|
|
|
mock_cli.create_job.assert_called_once()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_with_detailed_flag(self, mock_cli_class):
|
|
"""Test main function with detailed flag."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'list-kg-jobs-and-exit', '--detailed']):
|
|
main()
|
|
|
|
mock_cli.print_server_jobs.assert_called_once_with(True)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_with_custom_server_url(self, mock_cli_class):
|
|
"""Test main function with custom server URL."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'list-kg-jobs-and-exit', '--server_url', 'http://custom:8000/api/']):
|
|
main()
|
|
|
|
mock_cli_class.assert_called_once()
|
|
call_args = mock_cli_class.call_args
|
|
assert call_args[1]['base_url'] == 'http://custom:8000/api/'
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_with_username_and_token(self, mock_cli_class):
|
|
"""Test main function with username and token."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'list-kg-jobs-and-exit', '--username', 'testuser', '--token', 'testtoken']):
|
|
main()
|
|
|
|
mock_cli_class.assert_called_once()
|
|
call_args = mock_cli_class.call_args
|
|
assert call_args[1]['username'] == 'testuser'
|
|
assert call_args[1]['token'] == 'testtoken'
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_with_custom_input_output_prefixes(self, mock_cli_class):
|
|
"""Test main function with custom input and output prefixes."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', [
|
|
'cswarm_text_to_kg_client.py',
|
|
'--action', 'list-kg-jobs-and-exit',
|
|
'--input_prefix', '/custom/input',
|
|
'--output_prefix', '/custom/output'
|
|
]):
|
|
main()
|
|
|
|
mock_cli_class.assert_called_once()
|
|
call_args = mock_cli_class.call_args
|
|
assert call_args[1]['input_prefix'] == '/custom/input'
|
|
assert call_args[1]['output_prefix'] == '/custom/output'
|