CoreRasurae
806 lines
34 KiB
Python
806 lines
34 KiB
Python
"""
|
|
Unit tests for cswarm_text_to_kg_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_text_to_kg_client import TextoToKGCLI, main
|
|
from cleverswarm_python_client.libs.exceptions import ClientException
|
|
from cleverswarm_python_client.libs.file_type_enum import FileTypeAPI
|
|
from cleverswarm_python_client.libs.job_enums import JobStatus, JobType
|
|
|
|
|
|
class TestTextoToKGCLIInit:
|
|
"""Test cases for TextoToKGCLI initialization."""
|
|
|
|
def test_init_with_defaults(self):
|
|
"""Test initialization with default parameters."""
|
|
cli = TextoToKGCLI("http://localhost:8000/api/v0/")
|
|
|
|
assert cli._detailed is False
|
|
assert cli._input_prefix == Path("../..").resolve()
|
|
assert cli._output_prefix == Path("../..").resolve()
|
|
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 = TextoToKGCLI(
|
|
"http://test:8080/api/",
|
|
detailed=True,
|
|
input_prefix=temp_dir,
|
|
output_prefix=temp_dir,
|
|
username="testuser",
|
|
token="testtoken"
|
|
)
|
|
|
|
assert cli._detailed is True
|
|
assert cli._input_prefix == Path(temp_dir).resolve()
|
|
assert cli._output_prefix == Path(temp_dir).resolve()
|
|
assert cli._client is not None
|
|
|
|
def test_init_with_none_prefixes(self):
|
|
"""Test initialization with None prefixes."""
|
|
cli = TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix=None,
|
|
output_prefix=None
|
|
)
|
|
|
|
assert cli._input_prefix == Path("../..").resolve()
|
|
assert cli._output_prefix == Path("../..").resolve()
|
|
|
|
def test_init_invalid_input_prefix(self):
|
|
"""Test initialization with invalid input prefix."""
|
|
with pytest.raises(ClientException, match="Input prefix does not exist"):
|
|
TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix="/nonexistent/path"
|
|
)
|
|
|
|
def test_init_creates_output_prefix(self):
|
|
"""Test that initialization creates output prefix if it doesn't exist."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_dir = Path(temp_dir) / "new_output_dir"
|
|
|
|
cli = TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix=temp_dir,
|
|
output_prefix=str(output_dir)
|
|
)
|
|
|
|
assert output_dir.exists()
|
|
assert output_dir.is_dir()
|
|
|
|
def test_init_output_prefix_creation_failure(self):
|
|
"""Test initialization when output prefix creation fails."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
# Create a file with the same name as the output directory
|
|
output_file = Path(temp_dir) / "output_file"
|
|
output_file.write_text("test")
|
|
|
|
# This test expects a FileExistsError when trying to create a directory
|
|
# where a file with the same name already exists
|
|
with pytest.raises(FileExistsError):
|
|
TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix=temp_dir,
|
|
output_prefix=str(output_file)
|
|
)
|
|
|
|
|
|
class TestTextoToKGCLIJobCreation:
|
|
"""Test cases for TextoToKGCLI job creation methods."""
|
|
|
|
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_success(self):
|
|
"""Test successful job creation."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as json_file:
|
|
json_file.write('{"ontology": "data"}')
|
|
json_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
with patch.object(self.cli._client, 'create_unstructured_to_kg_job') as mock_create:
|
|
mock_create.return_value = "test_job_123"
|
|
|
|
result = self.cli.create_job(
|
|
text_file.name,
|
|
json_file.name,
|
|
owl_file.name
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
mock_create.assert_called_once()
|
|
|
|
def test_create_job_with_wildcards(self):
|
|
"""Test job creation with wildcards."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as json_file:
|
|
json_file.write('{"ontology": "data"}')
|
|
json_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.wildcards') as wildcards_file:
|
|
wildcards_file.write('{"wildcards": "data"}')
|
|
wildcards_file.flush()
|
|
|
|
with patch.object(self.cli._client, 'create_unstructured_to_kg_wildcards_job') as mock_create:
|
|
mock_create.return_value = "test_job_123"
|
|
|
|
result = self.cli.create_job(
|
|
text_file.name,
|
|
json_file.name,
|
|
owl_file.name,
|
|
wildcards_file.name
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
mock_create.assert_called_once()
|
|
|
|
def test_create_job_with_force_filetype(self):
|
|
"""Test job creation with forced file type."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as json_file:
|
|
json_file.write('{"ontology": "data"}')
|
|
json_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
with patch.object(self.cli._client, 'create_unstructured_to_kg_job') as mock_create:
|
|
mock_create.return_value = "test_job_123"
|
|
|
|
result = self.cli.create_job(
|
|
text_file.name,
|
|
json_file.name,
|
|
owl_file.name,
|
|
force_filetype=FileTypeAPI.Markdown
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
# Check that force_filetype was passed
|
|
call_args = mock_create.call_args
|
|
assert call_args[1]['force_filetype'] == FileTypeAPI.Markdown
|
|
|
|
def test_create_job_invalid_text_file(self):
|
|
"""Test job creation with invalid text file."""
|
|
with pytest.raises(ClientException, match="Unstructured input text does not exist"):
|
|
self.cli.create_job(
|
|
"nonexistent.txt",
|
|
"test.json",
|
|
"test.owl"
|
|
)
|
|
|
|
def test_create_job_with_invalid_json_file(self):
|
|
"""Test job creation with invalid (non-existent) JSON file - should raise exception if JSON is provided."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
# If ontology_json is provided but file doesn't exist, should raise exception
|
|
with pytest.raises(ClientException, match="Specified optional Ontology JSON file does not exist"):
|
|
self.cli.create_job(
|
|
text_file.name,
|
|
"nonexistent.json", # Non-existent JSON file - should raise exception
|
|
owl_file.name
|
|
)
|
|
|
|
def test_create_job_without_json_file(self):
|
|
"""Test job creation without JSON file - should work since JSON is optional."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
with patch.object(self.cli._client, 'create_unstructured_to_kg_job') as mock_create:
|
|
mock_create.return_value = "test_job_123"
|
|
|
|
# Job creation should work without JSON file since it's optional
|
|
result = self.cli.create_job(
|
|
text_file.name,
|
|
None, # No JSON file provided - should work
|
|
owl_file.name
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
mock_create.assert_called_once()
|
|
# Verify that None was passed for ontology_file since it wasn't provided
|
|
call_args = mock_create.call_args
|
|
assert call_args[0][1] is None # ontology_file parameter should be None
|
|
|
|
def test_create_job_with_empty_json_file_string(self):
|
|
"""Test job creation with empty string for JSON file - should work since JSON is optional."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
with patch.object(self.cli._client, 'create_unstructured_to_kg_job') as mock_create:
|
|
mock_create.return_value = "test_job_123"
|
|
|
|
# Job creation should work with empty string for JSON file since it's optional
|
|
result = self.cli.create_job(
|
|
text_file.name,
|
|
"", # Empty string for JSON file - should work
|
|
owl_file.name
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
mock_create.assert_called_once()
|
|
# Verify that None was passed for ontology_file since empty string is falsy
|
|
call_args = mock_create.call_args
|
|
assert call_args[0][1] is None # ontology_file parameter should be None
|
|
|
|
def test_create_job_invalid_owl_file(self):
|
|
"""Test job creation with invalid OWL file."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as json_file:
|
|
json_file.write('{"ontology": "data"}')
|
|
json_file.flush()
|
|
|
|
with pytest.raises(ClientException, match="Input ontology OWL file does not exist"):
|
|
self.cli.create_job(
|
|
text_file.name,
|
|
json_file.name,
|
|
"nonexistent.owl"
|
|
)
|
|
|
|
def test_create_job_invalid_wildcards_file(self):
|
|
"""Test job creation with invalid wildcards file."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as json_file:
|
|
json_file.write('{"ontology": "data"}')
|
|
json_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
with pytest.raises(ClientException, match="Input wildcards file does not exist"):
|
|
self.cli.create_job(
|
|
text_file.name,
|
|
json_file.name,
|
|
owl_file.name,
|
|
"nonexistent.wildcards"
|
|
)
|
|
|
|
def test_create_job_none_wildcards(self):
|
|
"""Test job creation with None wildcards."""
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as text_file:
|
|
text_file.write("Test text content")
|
|
text_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as json_file:
|
|
json_file.write('{"ontology": "data"}')
|
|
json_file.flush()
|
|
|
|
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.owl') as owl_file:
|
|
owl_file.write('<rdf:RDF>test</rdf:RDF>')
|
|
owl_file.flush()
|
|
|
|
with patch.object(self.cli._client, 'create_unstructured_to_kg_job') as mock_create:
|
|
mock_create.return_value = "test_job_123"
|
|
|
|
result = self.cli.create_job(
|
|
text_file.name,
|
|
json_file.name,
|
|
owl_file.name,
|
|
None
|
|
)
|
|
|
|
assert result == "test_job_123"
|
|
mock_create.assert_called_once()
|
|
|
|
|
|
class TestTextoToKGCLIJobPolling:
|
|
"""Test cases for TextoToKGCLI job polling methods."""
|
|
|
|
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
|
|
)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_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_text_to_kg_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_text_to_kg_client.CleverSwarmClient')
|
|
def test_poll_job_to_completion_processing(self, mock_client_class):
|
|
"""Test job polling with processing job."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
mock_client.get_job_status.return_value = JobStatus.Processing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Completed
|
|
|
|
# Should not raise any exception
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_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_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
mock_client.get_job_status.return_value = JobStatus.ReadyForProcessing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Completed
|
|
|
|
# Should not raise any exception
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_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_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
mock_client.get_job_status.return_value = JobStatus.Created
|
|
|
|
with pytest.raises(ClientException, match="is not processing"):
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_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_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
mock_client.get_job_status.return_value = JobStatus.Processing
|
|
mock_client.poll_job_ready_or_failed.return_value = JobStatus.Failed
|
|
|
|
with pytest.raises(ClientException, match="finished without completing"):
|
|
self.cli.poll_job_to_completion("test_job_123")
|
|
|
|
|
|
class TestTextoToKGCLIJobDownload:
|
|
"""Test cases for TextoToKGCLI job download methods."""
|
|
|
|
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
|
|
)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.CleverSwarmClient')
|
|
def test_download_job_result_success(self, mock_client_class):
|
|
"""Test successful job result download."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = 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
|
|
|
|
# Should not raise any exception
|
|
self.cli.download_job_result("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.CleverSwarmClient')
|
|
def test_download_job_result_no_input_files(self, mock_client_class):
|
|
"""Test job result download with no input files."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = 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")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.CleverSwarmClient')
|
|
def test_download_job_result_multiple_input_files(self, mock_client_class):
|
|
"""Test job result download with multiple input files."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = 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")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.CleverSwarmClient')
|
|
def test_download_job_result_detailed_output(self, mock_client_class, capsys):
|
|
"""Test job result download with detailed output."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = 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
|
|
|
|
self.cli.download_job_result("test_job_123")
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Job result was downloaded to:" in captured.out
|
|
|
|
|
|
class TestTextoToKGCLIJobManagement:
|
|
"""Test cases for TextoToKGCLI job management methods."""
|
|
|
|
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
|
|
)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.CleverSwarmClient')
|
|
def test_list_server_jobs_success(self, mock_client_class):
|
|
"""Test successful server jobs listing."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
mock_jobs = [
|
|
{"id": "job1", "status": "Completed"},
|
|
{"id": "job2", "status": "Processing"}
|
|
]
|
|
mock_client.get_jobs_list.return_value = mock_jobs
|
|
|
|
result = self.cli.list_server_jobs()
|
|
assert result == mock_jobs
|
|
mock_client.get_jobs_list.assert_called_once_with(is_benchmark=False)
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.CleverSwarmClient')
|
|
def test_delete_server_job_success(self, mock_client_class):
|
|
"""Test successful server job deletion."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
mock_client.delete_job.return_value = True
|
|
|
|
# Should not raise any exception
|
|
self.cli.delete_server_job("test_job_123")
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.CleverSwarmClient')
|
|
def test_delete_server_job_failure(self, mock_client_class):
|
|
"""Test server job deletion failure."""
|
|
mock_client = Mock()
|
|
mock_client_class.return_value = mock_client
|
|
self.cli._client = mock_client
|
|
|
|
mock_client.delete_job.return_value = False
|
|
|
|
# Should not raise any exception, but should handle failure gracefully
|
|
self.cli.delete_server_job("test_job_123")
|
|
|
|
|
|
class TestTextoToKGCLIJobPrinting:
|
|
"""Test cases for TextoToKGCLI job printing methods."""
|
|
|
|
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
|
|
)
|
|
|
|
@patch.object(TextoToKGCLI, 'list_server_jobs')
|
|
def test_print_server_jobs_basic(self, mock_list_jobs, capsys):
|
|
"""Test printing server jobs in basic mode."""
|
|
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"
|
|
}
|
|
]
|
|
mock_list_jobs.return_value = mock_jobs
|
|
|
|
self.cli.print_server_jobs()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Job ID: job1" in captured.out
|
|
assert "Status: Completed" in captured.out
|
|
|
|
@patch.object(TextoToKGCLI, 'list_server_jobs')
|
|
def test_print_server_jobs_detailed(self, mock_list_jobs, capsys):
|
|
"""Test printing server jobs in detailed mode."""
|
|
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_list_jobs.return_value = mock_jobs
|
|
|
|
self.cli.print_server_jobs(is_detailed=True)
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Job ID: job1" in captured.out
|
|
assert "Unstructured text input files:" in captured.out
|
|
assert "Ontology JSON input files:" in captured.out
|
|
assert "Ontology OWL/XML input files:" in captured.out
|
|
|
|
@patch.object(TextoToKGCLI, 'list_server_jobs')
|
|
def test_print_server_jobs_with_wildcards(self, mock_list_jobs, capsys):
|
|
"""Test printing server jobs with wildcards."""
|
|
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_list_jobs.return_value = mock_jobs
|
|
|
|
self.cli.print_server_jobs(is_detailed=True)
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Wildcards input files:" in captured.out
|
|
|
|
|
|
class TestTextoToKGCLIMainFunction:
|
|
"""Test cases for TextoToKGCLI main function."""
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_list_jobs(self, mock_cli_class):
|
|
"""Test main function with list jobs 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_and_exit_missing_args(self, mock_cli_class):
|
|
"""Test main function with create and exit action missing required args."""
|
|
mock_cli = Mock()
|
|
mock_cli_class.return_value = mock_cli
|
|
|
|
with patch('sys.argv', ['cswarm_text_to_kg_client.py', '--action', 'create-and-exit']):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_create_download_kg_delete(self, mock_cli_class):
|
|
"""Test main function with create-download-kg-delete action."""
|
|
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-download-kg-delete',
|
|
'--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")
|
|
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_client_exception(self, mock_cli_class):
|
|
"""Test main function with client exception."""
|
|
mock_cli = Mock()
|
|
mock_cli.create_job.side_effect = ClientException("Test error")
|
|
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'
|
|
]):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
@patch('cleverswarm_python_client.cswarm_text_to_kg_client.TextoToKGCLI')
|
|
def test_main_unexpected_exception(self, mock_cli_class):
|
|
"""Test main function with unexpected exception."""
|
|
mock_cli = Mock()
|
|
mock_cli.create_job.side_effect = Exception("Unexpected error")
|
|
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'
|
|
]):
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
|
|
|
|
class TestTextoToKGCLIIntegration:
|
|
"""Integration test cases for TextoToKGCLI."""
|
|
|
|
def test_full_workflow_simulation(self):
|
|
"""Test a complete workflow simulation."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix=temp_dir,
|
|
output_prefix=temp_dir
|
|
)
|
|
|
|
# Test that client is properly initialized
|
|
assert cli._client is not None
|
|
assert cli._input_prefix == Path(temp_dir).resolve()
|
|
assert cli._output_prefix == Path(temp_dir).resolve()
|
|
|
|
def test_path_resolution(self):
|
|
"""Test path resolution functionality."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix=temp_dir,
|
|
output_prefix=temp_dir
|
|
)
|
|
|
|
# Test that paths are resolved correctly
|
|
assert cli._input_prefix.is_absolute()
|
|
assert cli._output_prefix.is_absolute()
|
|
|
|
def test_file_validation_integration(self):
|
|
"""Test file validation in integration context."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
cli = TextoToKGCLI(
|
|
"http://localhost:8000/api/v0/",
|
|
input_prefix=temp_dir,
|
|
output_prefix=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>')
|
|
|
|
# Test that files are found correctly
|
|
assert text_file.exists()
|
|
assert json_file.exists()
|
|
assert owl_file.exists()
|
|
|
|
# Test job creation with real files
|
|
with patch.object(cli._client, 'create_unstructured_to_kg_job') as mock_create:
|
|
mock_create.return_value = "test_job_123"
|
|
|
|
result = cli.create_job(
|
|
str(text_file),
|
|
str(json_file),
|
|
str(owl_file)
|
|
)
|
|
|
|
assert result == "test_job_123"
|