""" Tests for the Atlas Record Cleaner Report application. """ from django.test import TestCase, Client from django.urls import reverse from django.core.cache import cache from unittest.mock import patch, MagicMock from datetime import datetime from .services.data_mapper import ( map_date, map_coordinate_uncertainty, map_occurrence_to_record_cleaner, merge_validation_and_verification_results ) from .services.atlas_service import parse_search_url, build_atlas_search_url from .services.record_cleaner_service import RecordCleanerClient, generate_summary_report class DataMapperTests(TestCase): """Tests for data mapping functions.""" def test_map_date_valid_timestamp(self): """Test mapping a valid UTC timestamp to date string.""" occurrence = {'eventDate': 1704067200000} # 2024-01-01 00:00:00 UTC result = map_date(occurrence) # Result will depend on timezone, but should be in DD/MM/YYYY format self.assertRegex(result, r'\d{2}/\d{2}/\d{4}') def test_map_date_missing(self): """Test mapping when eventDate is missing.""" occurrence = {} result = map_date(occurrence) self.assertEqual(result, '') def test_map_date_invalid(self): """Test mapping an invalid date.""" occurrence = {'eventDate': 'invalid'} result = map_date(occurrence) self.assertEqual(result, 'invalid') def test_map_coordinate_uncertainty_none(self): """Test mapping None uncertainty.""" result = map_coordinate_uncertainty(None) self.assertEqual(result, 1) def test_map_coordinate_uncertainty_zero(self): """Test mapping zero uncertainty.""" result = map_coordinate_uncertainty(0) self.assertEqual(result, 1) def test_map_coordinate_uncertainty_small(self): """Test mapping small uncertainty values.""" self.assertEqual(map_coordinate_uncertainty(5), 10) self.assertEqual(map_coordinate_uncertainty(9), 10) def test_map_coordinate_uncertainty_exact_power_of_ten(self): """Test mapping exact powers of ten.""" self.assertEqual(map_coordinate_uncertainty(1), 1) self.assertEqual(map_coordinate_uncertainty(10), 10) self.assertEqual(map_coordinate_uncertainty(100), 100) def test_map_coordinate_uncertainty_between_powers(self): """Test mapping values between powers of ten.""" self.assertEqual(map_coordinate_uncertainty(15), 100) self.assertEqual(map_coordinate_uncertainty(150), 1000) self.assertEqual(map_coordinate_uncertainty(1500), 10000) def test_map_occurrence_to_record_cleaner_gridref_default(self): """Grid references should be used by default when present.""" occurrence = { 'taxonConceptID': 'NBNSYS0000008319', 'scientificName': 'Adalia bipunctata', 'eventDate': 1704067200000, 'gridReference': 'SK1234', 'decimalLatitude': 52.5, 'decimalLongitude': -1.5, 'coordinateUncertaintyInMeters': 100, 'lifeStage': 'adult' } result = map_occurrence_to_record_cleaner(occurrence, 1) self.assertEqual(result['id'], 1) self.assertEqual(result['tvk'], 'NBNSYS0000008319') self.assertEqual(result['name'], 'Adalia bipunctata') self.assertRegex(result['date'], r'\d{2}/\d{2}/\d{4}') self.assertEqual(result['sref'], {'srid': 27700, 'gridref': 'SK1234'}) self.assertEqual(result['stage'], 'adult') def test_map_occurrence_to_record_cleaner_latlon_fallback(self): """Lat/lon fallback should be used when no grid reference is supplied.""" occurrence = { 'scientificName': 'Test species', 'decimalLatitude': 52.0, 'decimalLongitude': -1.0, } result = map_occurrence_to_record_cleaner(occurrence, 1) self.assertEqual(result['id'], 1) self.assertEqual(result['name'], 'Test species') self.assertNotIn('tvk', result) self.assertEqual(result['sref']['srid'], 4326) self.assertEqual(result['sref']['latitude'], 52.0) self.assertEqual(result['sref']['longitude'], -1.0) self.assertEqual(result['sref']['accuracy'], 1) # Default def test_merge_validation_and_verification_results(self): """Test merging occurrence with validation and verification results.""" occurrence = { 'occurrenceID': 'OCC123', 'taxonConceptLSID': 'TVK123', 'scientificName': 'Test species', 'eventDate': 1704067200000, 'decimalLatitude': 52.0, 'decimalLongitude': -1.0, } validated = { 'id': 1, 'result': 'pass', 'messages': [], 'preferred_tvk': 'TVK123' } verified = { 'id': 1, 'result': 'pass', 'messages': ['Rules run: tenkm'], 'organism_key': 'ORG123', 'id_difficulty': 1 } result = merge_validation_and_verification_results(occurrence, validated, verified) self.assertEqual(result['occurrenceID'], 'OCC123') self.assertEqual(result['validation_result'], 'pass') self.assertEqual(result['verification_result'], 'pass') self.assertEqual(result['preferred_tvk'], 'TVK123') self.assertEqual(result['id_difficulty'], 1) self.assertNotIn('organism_key', result) class AtlasServiceTests(TestCase): """Tests for NBN Atlas service functions.""" def test_parse_search_url_simple(self): """Test parsing a simple search URL.""" url = 'https://records.nbnatlas.org/occurrences/search?q=genus:Vulpes' result = parse_search_url(url) self.assertEqual(result['q'], 'genus:Vulpes') self.assertEqual(result['fq'], []) def test_parse_search_url_with_filters(self): """Test parsing a URL with multiple fq parameters.""" url = 'https://records.nbnatlas.org/occurrences/search?q=*:*&fq=genus:Vulpes&fq=year:2023' result = parse_search_url(url) self.assertEqual(result['q'], '*:*') self.assertIn('genus:Vulpes', result['fq']) self.assertIn('year:2023', result['fq']) def test_build_atlas_search_url_dataset(self): """Test building URL for dataset search.""" url = build_atlas_search_url(data_resource_uid='dr123', page_size=100) self.assertIn('data_resource_uid%3Adr123', url) # URL encoded self.assertIn('pageSize=100', url) self.assertIn('q=%2A%3A%2A', url) # URL encoded *:* def test_build_atlas_search_url_from_url(self): """Test building URL from user-provided search URL.""" search_url = 'https://records.nbnatlas.org/occurrences/search?q=genus:Vulpes&fq=year:2023' url = build_atlas_search_url(search_url=search_url, page_size=100) self.assertIn('q=genus%3AVulpes', url) self.assertIn('fq=year%3A2023', url) self.assertIn('pageSize=100', url) class RecordCleanerServiceTests(TestCase): """Tests for Record Cleaner service.""" @patch('atlas_record_cleaner_report.services.record_cleaner_service.requests.post') def test_get_token_success(self, mock_post): """Test successful token retrieval.""" mock_response = MagicMock() mock_response.json.return_value = {'access_token': 'test_token_123'} mock_post.return_value = mock_response client = RecordCleanerClient('user', 'pass') token = client._get_token() self.assertEqual(token, 'test_token_123') self.assertIsNotNone(client.token_expires_at) @patch('atlas_record_cleaner_report.services.record_cleaner_service.requests.post') def test_validate_records(self, mock_post): """Test record validation.""" # Mock token response token_response = MagicMock() token_response.json.return_value = {'access_token': 'test_token'} # Mock validate response validate_response = MagicMock() validate_response.json.return_value = [ {'id': 1, 'result': 'pass', 'messages': []} ] mock_post.side_effect = [token_response, validate_response] client = RecordCleanerClient('user', 'pass') records = [{'id': 1, 'name': 'Test', 'date': '01/01/2024', 'sref': {'srid': 4326}}] result = client.validate(records) self.assertEqual(len(result), 1) self.assertEqual(result[0]['result'], 'pass') def test_generate_summary_report(self): """Test summary report generation.""" validated = [ {'result': 'pass'}, {'result': 'pass'}, {'result': 'fail'}, ] verified = [ {'result': 'pass'}, {'result': 'warn'}, ] summary = generate_summary_report(validated, verified) self.assertEqual(summary['validation']['total'], 3) self.assertEqual(summary['validation']['pass'], 2) self.assertEqual(summary['validation']['fail'], 1) self.assertEqual(summary['verification']['total'], 2) self.assertEqual(summary['verification']['pass'], 1) self.assertEqual(summary['verification']['warn'], 1) self.assertAlmostEqual(summary['validation_pass_rate'], 66.7, places=1) self.assertEqual(summary['not_verified_count'], 1) class ViewTests(TestCase): """Tests for views.""" def setUp(self): """Set up test client and clear cache.""" self.client = Client() cache.clear() def test_index_view_get(self): """Test GET request to index view.""" with patch('atlas_record_cleaner_report.views.fetch_data_resources') as mock_fetch: mock_fetch.return_value = [ {'uid': 'dr1', 'name': 'Dataset 1', 'resourceType': 'records'}, {'uid': 'dr2', 'name': 'Dataset 2', 'resourceType': 'records'}, ] response = self.client.get(reverse('atlas_record_cleaner_report:index')) self.assertEqual(response.status_code, 200) self.assertContains(response, 'NBN Atlas Record Cleaner') self.assertContains(response, 'Dataset 1') self.assertContains(response, 'Dataset 2') def test_index_view_caches_datasets(self): """Test that datasets are cached.""" with patch('atlas_record_cleaner_report.views.fetch_data_resources') as mock_fetch: mock_fetch.return_value = [{'uid': 'dr1', 'name': 'Dataset 1', 'resourceType': 'records'}] # First request should fetch self.client.get(reverse('atlas_record_cleaner_report:index')) self.assertEqual(mock_fetch.call_count, 1) # Second request should use cache self.client.get(reverse('atlas_record_cleaner_report:index')) self.assertEqual(mock_fetch.call_count, 1) # Still 1, not called again def test_download_csv_no_data(self): """Test downloading CSV when no report data exists.""" response = self.client.get(reverse('atlas_record_cleaner_report:download_csv')) self.assertEqual(response.status_code, 404) def test_download_csv_with_data(self): """Test downloading CSV with report data in session.""" session = self.client.session session['report_data'] = [ { 'occurrence_id': 'OCC1', 'scientific_name': 'Test species', 'validation_result': 'pass', 'verification_result': 'pass' } ] session.save() response = self.client.get(reverse('atlas_record_cleaner_report:download_csv')) self.assertEqual(response.status_code, 200) self.assertEqual(response['Content-Type'], 'text/csv') self.assertIn('attachment', response['Content-Disposition']) self.assertIn(b'occurrence_id', response.content) self.assertIn(b'Test species', response.content)