import os import tempfile import unittest import shutil import pytest # Skip entire module - depends on obsolete owilix.core.ui pytestmark = pytest.mark.skip(reason="Depends on obsolete owilix.core.ui module - needs refactoring") # Import setup function and utilities from the main functional test file from tests.owilix.cmd.functional_test import setup from owilix.cmd import RemoteCommands, LocalCommands, QueryCommands # TODO: NOT TESTED class GraphCommandTests(unittest.TestCase): """Test suite specifically for the graph command with all update modes""" def setUp(self): """Set up test context using the shared setup function""" self.temp_dir = tempfile.mkdtemp() self.ctx = setup() self.test_dataset_id = "up/id=25022e2c-3ab6-11f0-9249-0242ac140003" def tearDown(self): """Clean up temporary directory""" if os.path.exists(self.temp_dir): shutil.rmtree(self.temp_dir, ignore_errors=True) def _ensure_dataset_available(self): """Helper method to pull test dataset and verify it's available locally""" # Pull dataset if not already local _r = RemoteCommands(self.ctx['OWI'], **self.ctx).do("pull", self.test_dataset_id) self.assertTrue(_r[0]["success"], f"Failed to pull dataset {self.test_dataset_id}") self.assertTrue(_r[1] and len(_r[1]) == 1, f"Expected exactly 1 dataset, got {len(_r[1]) if _r[1] else 0}") # Verify dataset exists locally _l = LocalCommands(self.ctx['OWI'], **self.ctx).do("ls", self.test_dataset_id) self.assertTrue(_l[0]["success"], f"Failed to list local dataset {self.test_dataset_id}") self.assertTrue(_l[1] and len(_l[1]) == 1, f"Dataset not found locally: {self.test_dataset_id}") return _l[1][0] # Return dataset object def test_graph_update_mode_none(self): """Test graph command with update_mode=None (aggregated processing)""" # Ensure test dataset is available dataset = self._ensure_dataset_available() # Create temporary directory for graph output graph_output_dir = tempfile.mkdtemp(prefix="graph_test_none_") try: # Test graph command with update_mode=None (default) _g = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="test_graph", db_location=graph_output_dir, update_mode=None, # Aggregated mode batch_size=1000, pq_batch_size=1, files="**/*.parquet", verbose=True ) # Verify command succeeded self.assertTrue(_g[0]["success"], f"Graph command failed: {_g[0].get('msg', 'Unknown error')}") # Verify output files exist in specified directory expected_files = [ os.path.join(graph_output_dir, "test_graph.duckdb"), os.path.join(graph_output_dir, "test_graph_hosts.parquet"), os.path.join(graph_output_dir, "test_graph_domains.parquet") ] for expected_file in expected_files: self.assertTrue(os.path.exists(expected_file), f"Expected output file not found: {expected_file}") self.assertGreater(os.path.getsize(expected_file), 0, f"Output file is empty: {expected_file}") # Verify message indicates aggregated processing self.assertIn("Graph analysis completed", _g[0]["msg"]) finally: # Clean up temporary directory if os.path.exists(graph_output_dir): shutil.rmtree(graph_output_dir, ignore_errors=True) def test_graph_update_mode_create(self): """Test graph command with update_mode='create' (per-dataset, create only if not exists)""" # Ensure test dataset is available dataset = self._ensure_dataset_available() # Test graph command with update_mode='create' _g1 = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="test_graph_create", update_mode="create", # Create mode batch_size=1000, pq_batch_size=1, files="**/*.parquet", verbose=True ) # Verify first run succeeded self.assertTrue(_g1[0]["success"], f"First graph create command failed: {_g1[0].get('msg', 'Unknown error')}") self.assertIn("processed", _g1[0]["msg"].lower()) # Test second run - should skip because files already exist _g2 = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="test_graph_create", update_mode="create", # Create mode again batch_size=1000, pq_batch_size=1, files="**/*.parquet", verbose=True ) # Second run should succeed but might skip processing or process fewer datasets self.assertTrue(_g2[0]["success"], f"Second graph create command failed: {_g2[0].get('msg', 'Unknown error')}") def test_graph_update_mode_update(self): """Test graph command with update_mode='update' (per-dataset, always overwrite)""" # Ensure test dataset is available dataset = self._ensure_dataset_available() # Test graph command with update_mode='update' - first run _g1 = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="test_graph_update", update_mode="update", # Update mode batch_size=1000, pq_batch_size=1, files="**/*.parquet", verbose=True ) # Verify first run succeeded self.assertTrue(_g1[0]["success"], f"First graph update command failed: {_g1[0].get('msg', 'Unknown error')}") self.assertIn("processed", _g1[0]["msg"].lower()) # Test second run - should process again and overwrite _g2 = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="test_graph_update", update_mode="update", # Update mode again batch_size=1000, pq_batch_size=1, files="**/*.parquet", verbose=True ) # Second run should also succeed and process self.assertTrue(_g2[0]["success"], f"Second graph update command failed: {_g2[0].get('msg', 'Unknown error')}") self.assertIn("processed", _g2[0]["msg"].lower()) def test_graph_parameter_validation(self): """Test graph command parameter validation and edge cases""" # Test with non-existent dataset specifier _g = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", "all/id=non-existent-dataset-id", update_mode=None ) # Should handle gracefully - either fail or return no results # We don't assert success here since it depends on implementation details # Test with valid dataset but different parameters dataset = self._ensure_dataset_available() _g = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="test_small_batch", update_mode=None, batch_size=10, # Small batch size pq_batch_size=1, files="**/*.parquet" ) # Should still work with small batch size if dataset has compatible data # The exact result depends on the dataset content def test_graph_with_multiple_datasets(self): """Test graph command with multiple datasets using aggregated mode""" # Ensure our test dataset is available dataset = self._ensure_dataset_available() # Test aggregated mode with wildcard pattern (should include our test dataset) _g = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", "all", # Process all available local datasets outfile="test_multi", update_mode=None, # Aggregated mode batch_size=1000, pq_batch_size=1, files="**/*.parquet", verbose=True ) # Command should complete successfully if there's any compatible data # At minimum, it should not crash and should handle the case gracefully def test_graph_output_file_validation(self): """Test that graph command generates expected output files with correct structure""" # Ensure test dataset is available dataset = self._ensure_dataset_available() # Create temporary directory for graph output graph_output_dir = tempfile.mkdtemp(prefix="graph_output_validation_") try: # Run graph command _g = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="validation_test", db_location=graph_output_dir, update_mode=None, batch_size=500, pq_batch_size=1, files="**/*.parquet" ) if _g[0]["success"]: # Verify DuckDB file structure (basic check) duckdb_file = os.path.join(graph_output_dir, "validation_test.duckdb") if os.path.exists(duckdb_file): self.assertGreater(os.path.getsize(duckdb_file), 0, "DuckDB file should not be empty") # Verify parquet files can be read (basic structure check) import pandas as pd host_file = os.path.join(graph_output_dir, "validation_test_hosts.parquet") domain_file = os.path.join(graph_output_dir, "validation_test_domains.parquet") if os.path.exists(host_file): try: host_df = pd.read_parquet(host_file) # Check that expected columns exist expected_columns = ['node', 'pagerank', 'in_degree', 'out_degree'] for col in expected_columns: self.assertIn(col, host_df.columns, f"Expected column '{col}' not found in hosts file") except Exception as e: self.fail(f"Failed to read hosts parquet file: {e}") if os.path.exists(domain_file): try: domain_df = pd.read_parquet(domain_file) # Check that expected columns exist expected_columns = ['node', 'pagerank', 'in_degree', 'out_degree'] for col in expected_columns: self.assertIn(col, domain_df.columns, f"Expected column '{col}' not found in domains file") except Exception as e: self.fail(f"Failed to read domains parquet file: {e}") finally: # Clean up if os.path.exists(graph_output_dir): shutil.rmtree(graph_output_dir, ignore_errors=True) def test_graph_report_integration(self): """Test that graph command output can be used by report command""" # Ensure test dataset is available dataset = self._ensure_dataset_available() # First run graph command to generate stats _g = QueryCommands(self.ctx['OWI'], **self.ctx).do( "graph", self.test_dataset_id, outfile="integration_test", update_mode="create", # Store in dataset stats batch_size=1000, pq_batch_size=1, files="**/*.parquet" ) if _g[0]["success"]: # Try to run report command (assuming it exists) try: _r = QueryCommands(self.ctx['OWI'], **self.ctx).do( "report", self.test_dataset_id, outfile="integration_test_report.html", topk=100, ranking_metric="pagerank", update_mode="create" ) # If report command exists and works, it should succeed # This test verifies the integration between graph and report commands except AttributeError: # Report command might not be implemented yet self.skipTest("Report command not available for integration testing") class GraphCommandIRODSTests(GraphCommandTests): """Graph command tests specifically for IRODS-only configuration""" def setUp(self): """Set up test context using IRODS-only setup""" self.temp_dir = tempfile.mkdtemp() self.ctx = setup(irods_only=True) # Use IRODS-only configuration self.test_dataset_id = "all/id=a962a1bb-3a65-4c24-954c-a96c99d5e95e" # Inherit all tests from parent class # Add IRODS-specific tests here if needed if __name__ == "__main__": # Run only the graph command tests unittest.main()