diff --git a/django_project/monitor/tests/test_observations.py b/django_project/monitor/tests/test_observations.py index edd006da3..9a08895ea 100644 --- a/django_project/monitor/tests/test_observations.py +++ b/django_project/monitor/tests/test_observations.py @@ -15,535 +15,545 @@ from minisass_authentication.models import UserProfile from minisass_authentication.tests.factories import UserFactory from monitor.models import ( - Observations, Sites, SiteImage, ObservationPestImage, Pest + Observations, Sites, SiteImage, ObservationPestImage, Pest ) @override_settings( - MINIO_ENDPOINT='http://some-endpoint', - MINIO_ACCESS_KEY='MINIO_ACCESS_KEY', - MINIO_SECRET_KEY='MINIO_SECRET_KEY' + MINIO_ENDPOINT='http://some-endpoint', + MINIO_ACCESS_KEY='MINIO_ACCESS_KEY', + MINIO_SECRET_KEY='MINIO_SECRET_KEY' ) class BaseObservationsModelTest(TestCase): - """ - Base Test for Observation Test. - """ - def image_field(self, name): - """Return image field with name.""" - image_path = os.path.join( - os.path.abspath(os.path.dirname(__file__)), 'test_image.png' - ) - return SimpleUploadedFile( - name=name, - content=open(image_path, 'rb').read(), - content_type='image/jpeg' - ) - - def setUp(self): - self.s3_client_patch = patch('minisass.utils.boto3.client') - self.s3_client_patch.start() - - self.user = User.objects.create_user( - username='testuser', - password='testpassword', - first_name='First', - last_name='Last', - email='test@gmail.com' - ) - self.profile = UserProfile.objects.get_or_create(user=self.user) - self.client = APIClient() - self.client.force_authenticate(user=self.user) - - self.site = Sites.objects.create( - site_name='Test Site', - river_name='Test River', - the_geom=Point(0, 0), - user=self.user - ) - - # Site images - self.site_image_1 = SiteImage.objects.create( - site=self.site, - image=self.image_field('site_1.jpg') - ) - self.site_image_1 = SiteImage.objects.create( - site=self.site, - image=self.image_field('site_2.jpg') - ) - - self.observation = Observations.objects.create( - user=self.user, - flatworms=True, - worms=True, - leeches=False, - crabs_shrimps=False, - site=self.site, - comment='test_comment', - score=4.5, - obs_date=date(2023, 12, 3), - flag='clean', - water_clarity=7.5, - water_temp=25.0, - ph=7.0, - diss_oxygen=8.5, - diss_oxygen_unit='mgl', - elec_cond=15.0, - elec_cond_unit='S/m' - ) - - self.observation_2 = Observations.objects.create( - user=self.user, - flatworms=False, - worms=True, - leeches=True, - crabs_shrimps=False, - site=self.site, - comment='test_comment', - score=2, - obs_date=date(2023, 12, 7), - flag='clean', - water_clarity=3, - water_temp=2, - ph=1.0, - diss_oxygen=1.5, - diss_oxygen_unit='mgl', - elec_cond=1.0, - elec_cond_unit='S/m' - ) - - self.flatworms, _ = GroupScores.objects.get_or_create(name='Flatworms', sensitivity_score=2, db_field='flatworms') - self.bugs_beetles, _ = GroupScores.objects.get_or_create(name='Bugs beetles', sensitivity_score=2, db_field='bugs_beetles') - self.worms, _ = GroupScores.objects.get_or_create(name='Worms', sensitivity_score=2, db_field='worms') - self.leeches, _ = GroupScores.objects.get_or_create(name='Leeches', sensitivity_score=2, db_field='leeches') - - self.pest_image_1 = ObservationPestImage.objects.create( - observation=self.observation, - group=self.flatworms, - image=self.image_field('flatworms_1.jpg') - ) - - self.pest_image_2 = ObservationPestImage.objects.create( - observation=self.observation, - group=self.worms, - image=self.image_field('worms_1.jpg') - ) - - self.pest_image_3 = ObservationPestImage.objects.create( - observation=self.observation_2, - group=self.leeches, - image=self.image_field('leeches_1.jpg') - ) + """ + Base Test for Observation Test. + """ + def image_field(self, name, file_name): + """Return image field with name and file_name.""" + image_path = os.path.join( + os.path.abspath(os.path.dirname(__file__)), 'test_image.png' + ) + + # Determine content type based on file extension + extension = os.path.splitext(file_name)[1].lower() + if extension == '.png': + content_type = 'image/png' + elif extension in ['.jpg', '.jpeg']: + content_type = 'image/jpeg' + else: + raise ValueError(f"Unsupported file extension: {extension}") + + return SimpleUploadedFile( + name=name, + content=open(image_path, 'rb').read(), + content_type=content_type + ) + + def setUp(self): + self.s3_client_patch = patch('minisass.utils.boto3.client') + self.s3_client_patch.start() + + self.user = User.objects.create_user( + username='testuser', + password='testpassword', + first_name='First', + last_name='Last', + email='test@gmail.com' + ) + self.profile = UserProfile.objects.get_or_create(user=self.user) + self.client = APIClient() + self.client.force_authenticate(user=self.user) + + self.site = Sites.objects.create( + site_name='Test Site', + river_name='Test River', + the_geom=Point(0, 0), + user=self.user + ) + + # Site images + self.site_image_1 = SiteImage.objects.create( + site=self.site, + image=self.image_field('site_1.jpg') + ) + self.site_image_1 = SiteImage.objects.create( + site=self.site, + image=self.image_field('site_2.jpg') + ) + + self.observation = Observations.objects.create( + user=self.user, + flatworms=True, + worms=True, + leeches=False, + crabs_shrimps=False, + site=self.site, + comment='test_comment', + score=4.5, + obs_date=date(2023, 12, 3), + flag='clean', + water_clarity=7.5, + water_temp=25.0, + ph=7.0, + diss_oxygen=8.5, + diss_oxygen_unit='mgl', + elec_cond=15.0, + elec_cond_unit='S/m' + ) + + self.observation_2 = Observations.objects.create( + user=self.user, + flatworms=False, + worms=True, + leeches=True, + crabs_shrimps=False, + site=self.site, + comment='test_comment', + score=2, + obs_date=date(2023, 12, 7), + flag='clean', + water_clarity=3, + water_temp=2, + ph=1.0, + diss_oxygen=1.5, + diss_oxygen_unit='mgl', + elec_cond=1.0, + elec_cond_unit='S/m' + ) + + self.flatworms, _ = GroupScores.objects.get_or_create(name='Flatworms', sensitivity_score=2, db_field='flatworms') + self.bugs_beetles, _ = GroupScores.objects.get_or_create(name='Bugs beetles', sensitivity_score=2, db_field='bugs_beetles') + self.worms, _ = GroupScores.objects.get_or_create(name='Worms', sensitivity_score=2, db_field='worms') + self.leeches, _ = GroupScores.objects.get_or_create(name='Leeches', sensitivity_score=2, db_field='leeches') + + self.pest_image_1 = ObservationPestImage.objects.create( + observation=self.observation, + group=self.flatworms, + image=self.image_field('flatworms_1.jpg') + ) + + self.pest_image_2 = ObservationPestImage.objects.create( + observation=self.observation, + group=self.worms, + image=self.image_field('worms_1.jpg') + ) + + self.pest_image_3 = ObservationPestImage.objects.create( + observation=self.observation_2, + group=self.leeches, + image=self.image_field('leeches_1.jpg') + ) class ObservationsModelTest(BaseObservationsModelTest): - def test_observation_str_representation(self): - observation = Observations.objects.get(flatworms=True) - self.assertEqual( - str(observation), - f"{observation.obs_date}: {observation.site.site_name}" - ) - - def test_recent_observation_view(self): - url = reverse('recent-observation-list') - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - - # Ensure the response contains the necessary fields - self.assertContains(response, 'observation') - self.assertContains(response, 'site') - self.assertContains(response, 'username') - self.assertContains(response, 'organisation') - self.assertContains(response, 'time_stamp') - self.assertContains(response, 'score') - - def test_observation_retrieve_view(self): - url = reverse( - 'observation-details', kwargs={'pk': self.observation.gid} - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - response = response.json() - - # Ensure the response contains the necessary fields - self.assertEqual(response['sitename'], self.observation.site.site_name) - self.assertEqual( - response['rivername'], self.observation.site.river_name - ) - self.assertEqual( - response['sitedescription'], self.observation.site.description - ) - self.assertEqual( - response['rivercategory'], self.observation.site.river_cat - ) - self.assertEqual( - response['longitude'], self.observation.site.the_geom.x - ) - self.assertEqual( - response['latitude'], self.observation.site.the_geom.y - ) - self.assertEqual( - response['obs_date'], - datetime.strftime(self.observation.obs_date, "%Y-%m-%d") - ) - self.assertEqual( - response['collectorsname'], - f"{self.user.first_name} {self.user.last_name}" - ) - self.assertEqual(float(response['score']), self.observation.score) - self.assertEqual(response['flatworms'], self.observation.flatworms) - self.assertEqual(len(response['site']['images']), 2) - - def test_observation_delete_view(self): - observation = Observations.objects.first() - url = reverse( - 'observation-retrieve-update-delete', - kwargs={'pk': observation.gid} - ) - response = self.client.delete(url) - self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) - - # Check if the object is actually deleted from the database - with self.assertRaises(Observations.DoesNotExist): - Observations.objects.get(gid=observation.gid) - - def test_upload_pest_image(self): - - # Create a sample image file - image_file = SimpleUploadedFile("test_image.jpg", b"file_content", content_type="image/jpeg") - - # this resolves the user instance error when creating the site - self.client.login(email='test@gmail.com', password='testpassword') - - url = reverse('upload-pest-images') - - response = self.client.post( - url, - { - f'pest_image:{self.flatworms.id}': image_file, - 'observationId': 0, - 'siteId': 0, - 'create_site_or_observation': 'False' - }, - ) - - # Check the response status and content - self.assertEqual(response.status_code, 200) - self.assertIn('status', response.json()) - self.assertIn('observation_id', response.json()) - - observation_id = response.json()['observation_id'] - site_id = response.json()['site_id'] - pest_image_id = response.json()['pest_image_id'] - - - response = self.client.post( - url, - { - f'pest_image:{self.flatworms.id}': image_file, - 'observationId': observation_id, - 'siteId': site_id, - 'create_site_or_observation': 'False' - }, - ) - - observation_id = response.json()['observation_id'] - site_id = response.json()['site_id'] - pest_image_id = response.json()['pest_image_id'] - - url = reverse( - 'remove-pest-image', - kwargs={ - 'pk': pest_image_id, - 'observation_pk': observation_id - } - ) - - response = self.client.post(url) - - self.assertEqual(response.status_code, 200) - - def test_observation_image_list_view(self): - """Test observation images.""" - # List image for observation not exist - url = reverse( - 'observation-image-view-list', - kwargs={'observation_pk': 0} - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 404) - - # List image for observation 1 - url = reverse( - 'observation-image-view-list', - kwargs={'observation_pk': self.observation.gid} - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - response = response.json() - self.assertEqual(len(response), 2) - - # List image for observation 2 - url = reverse( - 'observation-image-view-list', - kwargs={'observation_pk': self.observation_2.gid} - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - response = response.json() - self.assertEqual(len(response), 1) - - def test_observation_image_detail_view(self): - """Test observation images.""" - - # Image does not exist - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': 0, - 'observation_pk': self.observation.gid - } - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 404) - - # Image 3 is not from observation 1 - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': self.pest_image_3.id, - 'observation_pk': self.observation.gid - } - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 404) - - # Image 1 is from observation 1 - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': self.pest_image_1.id, - 'observation_pk': self.observation.gid - } - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - response = response.json() - self.assertEqual(response['pest_id'], self.pest_image_1.group.id) - self.assertEqual(response['pest_name'], self.pest_image_1.group.name) - - def test_observation_image_delete_view(self): - """Test observation images.""" - - # Delete image with pk 0 - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': 0, - 'observation_pk': self.observation.gid - } - ) - self.client = APIClient() - response = self.client.delete(url) - self.assertEqual(response.status_code, 403) - - # Delete image that exist - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': self.pest_image_1.id, - 'observation_pk': self.observation.gid - } - ) - response = self.client.delete(url) - self.assertEqual(response.status_code, 403) - - # Delete image that exist but user is not creator of observation - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': self.pest_image_1.id, - 'observation_pk': self.observation.gid - } - ) - response = self.client.delete(url) - self.client.force_authenticate( - user=User.objects.create_user( - username='testuser_1', - password='testpassword_1', - first_name='First_1', - last_name='Last_1', - ) - ) - self.assertEqual(response.status_code, 403) - - # We get the pest_image using API - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': self.pest_image_1.id, - 'observation_pk': self.observation.gid - } - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 200) - - response = response.json() - self.assertEqual(response['pest_id'], self.pest_image_1.group.id) - self.assertEqual(response['pest_name'], self.pest_image_1.group.name) - - # Delete image that exist and user is creator of observation - self.client.force_authenticate( - user=self.user - ) - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': self.pest_image_1.id, - 'observation_pk': self.observation.gid - } - ) - response = self.client.delete(url) - self.assertEqual(response.status_code, 204) - - # We get the pest_image using API - url = reverse( - 'observation-image-view-detail', - kwargs={ - 'pk': self.pest_image_1.id, - 'observation_pk': self.observation.gid - } - ) - response = self.client.get(url) - self.assertEqual(response.status_code, 404) - - def test_observations_by_site_id(self): - self.client.force_authenticate( - user=self.user - ) - - # Make a GET request to the observations by site endpoint - url = reverse('observations-by-site', kwargs={'site_id': self.site.gid}) - - # Make a GET request to the observations by site endpoint - response = self.client.get(url) - self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response['Content-Type'], 'application/json') - - def test_observations_by_nonexistent_site_id(self): - self.client.login(email='test@gmail.com', password='testuserpassword') - - url = reverse('observations-by-site', kwargs={'site_id': 999}) - - # Make a GET request to the observations by site endpoint with a nonexistent site ID - response = self.client.get(url) - - # Check if the response status code is 404 Not Found - self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) - - def test_expert_auto_validated_observation(self): - """ - When observation is created by expert, it will automatically be validated. - All images will also valid. - """ - user = UserFactory.create() - user.userprofile.is_expert = True - user.userprofile.save() - - observation = Observations.objects.create( - user=user, - flatworms=True, - worms=True, - leeches=False, - crabs_shrimps=False, - site=self.site, - comment='test_comment', - score=4.5, - obs_date=date(2023, 12, 3), - flag='clean', - water_clarity=7.5, - water_temp=25.0, - ph=7.0, - diss_oxygen=8.5, - diss_oxygen_unit='mgl', - elec_cond=15.0, - elec_cond_unit='S/m' - ) - site_image_1 = SiteImage.objects.create( - site=observation.site, - image=self.image_field('site_2.jpg') - ) - pest_image_1 = ObservationPestImage.objects.create( - observation=observation, - group=self.bugs_beetles, - image=self.image_field('bugs_beetles_1.jpg') - ) - self.assertTrue(observation.is_validated) - self.assertTrue(pest_image_1.valid) - - # image is in 'clean' directory - self.assertTrue( - pest_image_1.image.name, - f'demo/observations/clean/bugs_beetles/{observation.site_id}/{observation.gid}/bugs_beetles_1.jpg' - ) - self.assertTrue( - site_image_1.image.name, - f'demo/sites/{observation.site_id}/site_2.jpg' - ) - - def test_novice_observation_dirty(self): - """ - When observation is created by novice, it will not automatically be validated. - All images will also invalid. - """ - self.assertFalse(self.observation.is_validated) - self.assertFalse(self.pest_image_1.valid) - - # image is in 'dirty' directory - self.assertTrue( - self.pest_image_1.image.name, - f'demo/observations/dirty/flatworms/{self.observation.site_id}/{self.observation.gid}/flatworms_1.jpg' - ) - self.assertTrue( - self.site_image_1.image.name, - f'demo/sites/{self.site.gid}/site_1.jpg' - ) - - # commenting this since it causes issue on deployment, even though it works on local and Github. - # def test_validate_image(self): - # """ - # Test validating image. Image path should be updated, and image file is moved. - # """ - # old_image_path = f'demo/observations/dirty/flatworms/{self.observation.site_id}/' \ - # f'{self.observation.gid}/flatworms_1.jpg' - # new_image_path = old_image_path.replace('/dirty/', '/clean/').replace('flatworms', 'worms') - # self.assertTrue( - # self.pest_image_1.image.name, - # old_image_path - # ) - # - # self.pest_image_1.valid = True - # self.pest_image_1.pest = self.worms - # self.pest_image_1.save() - # - # # image path in databse record is updated - # self.assertTrue( - # self.pest_image_1.image.name, - # new_image_path - # ) - # - # # Observation is validated - # self.assertTrue(self.pest_image_1.observation.is_validated) - # - # # Image file is moved to clean directory - # self.assertFalse(os.path.exists(os.path.join(settings.MINIO_ROOT, old_image_path))) - # self.assertTrue( - # os.path.exists(os.path.join(settings.MINIO_ROOT, new_image_path)) - # ) - - def tearDown(self): - """Tear down.""" - self.pest_image_1.delete() - self.pest_image_2.delete() - self.pest_image_3.delete() - self.site.delete() - self.s3_client_patch.stop() + def test_observation_str_representation(self): + observation = Observations.objects.get(flatworms=True) + self.assertEqual( + str(observation), + f"{observation.obs_date}: {observation.site.site_name}" + ) + + def test_recent_observation_view(self): + url = reverse('recent-observation-list') + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + + # Ensure the response contains the necessary fields + self.assertContains(response, 'observation') + self.assertContains(response, 'site') + self.assertContains(response, 'username') + self.assertContains(response, 'organisation') + self.assertContains(response, 'time_stamp') + self.assertContains(response, 'score') + + def test_observation_retrieve_view(self): + url = reverse( + 'observation-details', kwargs={'pk': self.observation.gid} + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + response = response.json() + + # Ensure the response contains the necessary fields + self.assertEqual(response['sitename'], self.observation.site.site_name) + self.assertEqual( + response['rivername'], self.observation.site.river_name + ) + self.assertEqual( + response['sitedescription'], self.observation.site.description + ) + self.assertEqual( + response['rivercategory'], self.observation.site.river_cat + ) + self.assertEqual( + response['longitude'], self.observation.site.the_geom.x + ) + self.assertEqual( + response['latitude'], self.observation.site.the_geom.y + ) + self.assertEqual( + response['obs_date'], + datetime.strftime(self.observation.obs_date, "%Y-%m-%d") + ) + self.assertEqual( + response['collectorsname'], + f"{self.user.first_name} {self.user.last_name}" + ) + self.assertEqual(float(response['score']), self.observation.score) + self.assertEqual(response['flatworms'], self.observation.flatworms) + self.assertEqual(len(response['site']['images']), 2) + + def test_observation_delete_view(self): + observation = Observations.objects.first() + url = reverse( + 'observation-retrieve-update-delete', + kwargs={'pk': observation.gid} + ) + response = self.client.delete(url) + self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) + + # Check if the object is actually deleted from the database + with self.assertRaises(Observations.DoesNotExist): + Observations.objects.get(gid=observation.gid) + + def test_upload_pest_image(self): + + # Create a sample image file + image_file = SimpleUploadedFile("test_image.jpg", b"file_content", content_type="image/jpeg") + + # this resolves the user instance error when creating the site + self.client.login(email='test@gmail.com', password='testpassword') + + url = reverse('upload-pest-images') + + response = self.client.post( + url, + { + f'pest_image:{self.flatworms.id}': image_file, + 'observationId': 0, + 'siteId': 0, + 'create_site_or_observation': 'False' + }, + ) + + # Check the response status and content + self.assertEqual(response.status_code, 200) + self.assertIn('status', response.json()) + self.assertIn('observation_id', response.json()) + + observation_id = response.json()['observation_id'] + site_id = response.json()['site_id'] + pest_image_id = response.json()['pest_image_id'] + + + response = self.client.post( + url, + { + f'pest_image:{self.flatworms.id}': image_file, + 'observationId': observation_id, + 'siteId': site_id, + 'create_site_or_observation': 'False' + }, + ) + + observation_id = response.json()['observation_id'] + site_id = response.json()['site_id'] + pest_image_id = response.json()['pest_image_id'] + + url = reverse( + 'remove-pest-image', + kwargs={ + 'pk': pest_image_id, + 'observation_pk': observation_id + } + ) + + response = self.client.post(url) + + self.assertEqual(response.status_code, 200) + + def test_observation_image_list_view(self): + """Test observation images.""" + # List image for observation not exist + url = reverse( + 'observation-image-view-list', + kwargs={'observation_pk': 0} + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 404) + + # List image for observation 1 + url = reverse( + 'observation-image-view-list', + kwargs={'observation_pk': self.observation.gid} + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + response = response.json() + self.assertEqual(len(response), 2) + + # List image for observation 2 + url = reverse( + 'observation-image-view-list', + kwargs={'observation_pk': self.observation_2.gid} + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + response = response.json() + self.assertEqual(len(response), 1) + + def test_observation_image_detail_view(self): + """Test observation images.""" + + # Image does not exist + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': 0, + 'observation_pk': self.observation.gid + } + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 404) + + # Image 3 is not from observation 1 + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': self.pest_image_3.id, + 'observation_pk': self.observation.gid + } + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 404) + + # Image 1 is from observation 1 + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': self.pest_image_1.id, + 'observation_pk': self.observation.gid + } + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + response = response.json() + self.assertEqual(response['pest_id'], self.pest_image_1.group.id) + self.assertEqual(response['pest_name'], self.pest_image_1.group.name) + + def test_observation_image_delete_view(self): + """Test observation images.""" + + # Delete image with pk 0 + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': 0, + 'observation_pk': self.observation.gid + } + ) + self.client = APIClient() + response = self.client.delete(url) + self.assertEqual(response.status_code, 403) + + # Delete image that exist + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': self.pest_image_1.id, + 'observation_pk': self.observation.gid + } + ) + response = self.client.delete(url) + self.assertEqual(response.status_code, 403) + + # Delete image that exist but user is not creator of observation + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': self.pest_image_1.id, + 'observation_pk': self.observation.gid + } + ) + response = self.client.delete(url) + self.client.force_authenticate( + user=User.objects.create_user( + username='testuser_1', + password='testpassword_1', + first_name='First_1', + last_name='Last_1', + ) + ) + self.assertEqual(response.status_code, 403) + + # We get the pest_image using API + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': self.pest_image_1.id, + 'observation_pk': self.observation.gid + } + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + + response = response.json() + self.assertEqual(response['pest_id'], self.pest_image_1.group.id) + self.assertEqual(response['pest_name'], self.pest_image_1.group.name) + + # Delete image that exist and user is creator of observation + self.client.force_authenticate( + user=self.user + ) + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': self.pest_image_1.id, + 'observation_pk': self.observation.gid + } + ) + response = self.client.delete(url) + self.assertEqual(response.status_code, 204) + + # We get the pest_image using API + url = reverse( + 'observation-image-view-detail', + kwargs={ + 'pk': self.pest_image_1.id, + 'observation_pk': self.observation.gid + } + ) + response = self.client.get(url) + self.assertEqual(response.status_code, 404) + + def test_observations_by_site_id(self): + self.client.force_authenticate( + user=self.user + ) + + # Make a GET request to the observations by site endpoint + url = reverse('observations-by-site', kwargs={'site_id': self.site.gid}) + + # Make a GET request to the observations by site endpoint + response = self.client.get(url) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response['Content-Type'], 'application/json') + + def test_observations_by_nonexistent_site_id(self): + self.client.login(email='test@gmail.com', password='testuserpassword') + + url = reverse('observations-by-site', kwargs={'site_id': 999}) + + # Make a GET request to the observations by site endpoint with a nonexistent site ID + response = self.client.get(url) + + # Check if the response status code is 404 Not Found + self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) + + def test_expert_auto_validated_observation(self): + """ + When observation is created by expert, it will automatically be validated. + All images will also valid. + """ + user = UserFactory.create() + user.userprofile.is_expert = True + user.userprofile.save() + + observation = Observations.objects.create( + user=user, + flatworms=True, + worms=True, + leeches=False, + crabs_shrimps=False, + site=self.site, + comment='test_comment', + score=4.5, + obs_date=date(2023, 12, 3), + flag='clean', + water_clarity=7.5, + water_temp=25.0, + ph=7.0, + diss_oxygen=8.5, + diss_oxygen_unit='mgl', + elec_cond=15.0, + elec_cond_unit='S/m' + ) + site_image_1 = SiteImage.objects.create( + site=observation.site, + image=self.image_field('site_2.jpg') + ) + pest_image_1 = ObservationPestImage.objects.create( + observation=observation, + group=self.bugs_beetles, + image=self.image_field('bugs_beetles_1.jpg') + ) + self.assertTrue(observation.is_validated) + self.assertTrue(pest_image_1.valid) + + # image is in 'clean' directory + self.assertTrue( + pest_image_1.image.name, + f'demo/observations/clean/bugs_beetles/{observation.site_id}/{observation.gid}/bugs_beetles_1.jpg' + ) + self.assertTrue( + site_image_1.image.name, + f'demo/sites/{observation.site_id}/site_2.jpg' + ) + + def test_novice_observation_dirty(self): + """ + When observation is created by novice, it will not automatically be validated. + All images will also invalid. + """ + self.assertFalse(self.observation.is_validated) + self.assertFalse(self.pest_image_1.valid) + + # image is in 'dirty' directory + self.assertTrue( + self.pest_image_1.image.name, + f'demo/observations/dirty/flatworms/{self.observation.site_id}/{self.observation.gid}/flatworms_1.jpg' + ) + self.assertTrue( + self.site_image_1.image.name, + f'demo/sites/{self.site.gid}/site_1.jpg' + ) + + # commenting this since it causes issue on deployment, even though it works on local and Github. + # def test_validate_image(self): + # """ + # Test validating image. Image path should be updated, and image file is moved. + # """ + # old_image_path = f'demo/observations/dirty/flatworms/{self.observation.site_id}/' \ + # f'{self.observation.gid}/flatworms_1.jpg' + # new_image_path = old_image_path.replace('/dirty/', '/clean/').replace('flatworms', 'worms') + # self.assertTrue( + # self.pest_image_1.image.name, + # old_image_path + # ) + # + # self.pest_image_1.valid = True + # self.pest_image_1.pest = self.worms + # self.pest_image_1.save() + # + # # image path in databse record is updated + # self.assertTrue( + # self.pest_image_1.image.name, + # new_image_path + # ) + # + # # Observation is validated + # self.assertTrue(self.pest_image_1.observation.is_validated) + # + # # Image file is moved to clean directory + # self.assertFalse(os.path.exists(os.path.join(settings.MINIO_ROOT, old_image_path))) + # self.assertTrue( + # os.path.exists(os.path.join(settings.MINIO_ROOT, new_image_path)) + # ) + + def tearDown(self): + """Tear down.""" + self.pest_image_1.delete() + self.pest_image_2.delete() + self.pest_image_3.delete() + self.site.delete() + self.s3_client_patch.stop() diff --git a/django_project/monitor/tests/test_sites.py b/django_project/monitor/tests/test_sites.py index 53434921b..d8e4f8444 100644 --- a/django_project/monitor/tests/test_sites.py +++ b/django_project/monitor/tests/test_sites.py @@ -15,209 +15,219 @@ class SitesListCreateViewTestCase(TestCase): - def image_field(self, name): - """Return image field with name.""" - image_path = os.path.join( - os.path.abspath(os.path.dirname(__file__)), 'test_image.png' - ) - return SimpleUploadedFile( - name=name, - content=open(image_path, 'rb').read(), - content_type='image/jpeg' - ) - - def setUp(self): - # Create a user for authentication - self.user = User.objects.create_user(username='testuser', password='testpassword', email='test@example.com') - self.site = Sites.objects.create( - site_name='Test Site', - river_name='Test River', - the_geom=Point(0, 0), - user=self.user - ) - - def test_multiple_image_upload(self): - client = APIClient() - client.force_authenticate(user=self.user) - - # Create multiple SimpleUploadedFile instances as a simulation of image files - image_files = [ - SimpleUploadedFile("image1.jpg", b"file_content", content_type="image/jpeg"), - SimpleUploadedFile("image2.jpg", b"file_content", content_type="image/jpeg"), - ] - - # Create a dictionary to simulate the request data - request_data = { - 'images': image_files, - } - - url = reverse('save_site_images', kwargs={'site_id': self.site.gid}) - - # Use the client to perform a POST request with the provided data - response = client.post(url, request_data, format='multipart') - - # Check if the response status code is 201 (HTTP_CREATED) - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - - # Check if the images were saved in the SiteImage model - saved_images = SiteImage.objects.filter(site=self.site) - - # Assert that the number of saved images matches the number of provided images - self.assertEqual(saved_images.count(), len(image_files)) - - def test_create_site_with_images(self): - client = APIClient() - client.force_authenticate(user=self.user) - - # Create a mock image - image = Image.new('RGB', (100, 100)) - image_file = BytesIO() - image.save(image_file, 'PNG') - image_file.name = 'test_image.png' - - # Prepare the request data - data = { - 'site_data': { - 'site_name': 'Test Site', - 'river_name': 'Test River', - 'description': 'Test Description', - 'river_cat': 'rocky', - 'longitude': 0, - 'latitude': 0, - }, - 'images': [image_file], - } - - # Make a POST request to create a site with images - url = reverse('sites-list-create') - response = client.post(url, data, format='json') - - # Check if the response status code is 201 Created - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - - # Check if the site and images are created - self.assertEqual(response.data['site_name'], 'Test Site') - self.assertEqual(response.data['river_name'], 'Test River') - self.assertEqual(response.data['description'], 'Test Description') - - # Ensure the images are attached to the site - site_id = response.data['gid'] - site = Sites.objects.get(gid=site_id) - self.site_image_1 = SiteImage.objects.create( - site=site, - image=self.image_field('site_1.jpg') - ) - images_count = SiteImage.objects.filter(site_id=site_id).count() - self.assertEqual(images_count, 1) - - def test_create_site_without_images(self): - client = APIClient() - client.force_authenticate(user=self.user) - - # Prepare the request data without images - data = { - 'site_data': { - 'site_name': 'Test Site', - 'river_name': 'Test River', - 'description': 'Test Description', - 'river_cat': 'rocky', - 'longitude': 0, - 'latitude': 0, - }, - 'images': [], - } - - # Make a POST request to create a site without images - url = reverse('sites-list-create') - response = client.post(url, data, format='json') - - # Check if the response status code is 201 Created - self.assertEqual(response.status_code, status.HTTP_201_CREATED) - - # Check if the site is created without images - self.assertEqual(response.data['site_name'], 'Test Site') - self.assertEqual(response.data['river_name'], 'Test River') - self.assertEqual(response.data['description'], 'Test Description') - - # Ensure no images are attached to the site - site_id = response.data['gid'] - images_count = SiteImage.objects.filter(site_id=site_id).count() - self.assertEqual(images_count, 0) - - def test_list_site_filter(self): - Sites.objects.create( - user=self.user, - site_name='Cape Town', - the_geom=Point(0, 0) - ) - Sites.objects.create( - user=self.user, - site_name='Limpopo', - the_geom=Point(1, 1) - ) - client = APIClient() - client.force_authenticate(user=self.user) - url = reverse('sites-list-create') - response = client.get(f'{url}?site_name=mpo', format='json') - self.assertEquals(len(response.json()), 1) - self.assertEquals(response.json()[0]['site_name'], 'Limpopo') + def image_field(self, name, file_name): + """Return image field with name and file_name.""" + image_path = os.path.join( + os.path.abspath(os.path.dirname(__file__)), 'test_image.png' + ) + + # Determine content type based on file extension + extension = os.path.splitext(file_name)[1].lower() + if extension == '.png': + content_type = 'image/png' + elif extension in ['.jpg', '.jpeg']: + content_type = 'image/jpeg' + else: + raise ValueError(f"Unsupported file extension: {extension}") + + return SimpleUploadedFile( + name=name, + content=open(image_path, 'rb').read(), + content_type=content_type + ) + + def setUp(self): + # Create a user for authentication + self.user = User.objects.create_user(username='testuser', password='testpassword', email='test@example.com') + self.site = Sites.objects.create( + site_name='Test Site', + river_name='Test River', + the_geom=Point(0, 0), + user=self.user + ) + + def test_multiple_image_upload(self): + client = APIClient() + client.force_authenticate(user=self.user) + + # Create multiple SimpleUploadedFile instances as a simulation of image files + image_files = [ + SimpleUploadedFile("image1.jpg", b"file_content", content_type="image/jpeg"), + SimpleUploadedFile("image2.jpg", b"file_content", content_type="image/jpeg"), + ] + + # Create a dictionary to simulate the request data + request_data = { + 'images': image_files, + } + + url = reverse('save_site_images', kwargs={'site_id': self.site.gid}) + + # Use the client to perform a POST request with the provided data + response = client.post(url, request_data, format='multipart') + + # Check if the response status code is 201 (HTTP_CREATED) + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + + # Check if the images were saved in the SiteImage model + saved_images = SiteImage.objects.filter(site=self.site) + + # Assert that the number of saved images matches the number of provided images + self.assertEqual(saved_images.count(), len(image_files)) + + def test_create_site_with_images(self): + client = APIClient() + client.force_authenticate(user=self.user) + + # Create a mock image + image = Image.new('RGB', (100, 100)) + image_file = BytesIO() + image.save(image_file, 'PNG') + image_file.name = 'test_image.png' + + # Prepare the request data + data = { + 'site_data': { + 'site_name': 'Test Site', + 'river_name': 'Test River', + 'description': 'Test Description', + 'river_cat': 'rocky', + 'longitude': 0, + 'latitude': 0, + }, + 'images': [image_file], + } + + # Make a POST request to create a site with images + url = reverse('sites-list-create') + response = client.post(url, data, format='json') + + # Check if the response status code is 201 Created + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + + # Check if the site and images are created + self.assertEqual(response.data['site_name'], 'Test Site') + self.assertEqual(response.data['river_name'], 'Test River') + self.assertEqual(response.data['description'], 'Test Description') + + # Ensure the images are attached to the site + site_id = response.data['gid'] + site = Sites.objects.get(gid=site_id) + self.site_image_1 = SiteImage.objects.create( + site=site, + image=self.image_field('site_1.jpg') + ) + images_count = SiteImage.objects.filter(site_id=site_id).count() + self.assertEqual(images_count, 1) + + def test_create_site_without_images(self): + client = APIClient() + client.force_authenticate(user=self.user) + + # Prepare the request data without images + data = { + 'site_data': { + 'site_name': 'Test Site', + 'river_name': 'Test River', + 'description': 'Test Description', + 'river_cat': 'rocky', + 'longitude': 0, + 'latitude': 0, + }, + 'images': [], + } + + # Make a POST request to create a site without images + url = reverse('sites-list-create') + response = client.post(url, data, format='json') + + # Check if the response status code is 201 Created + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + + # Check if the site is created without images + self.assertEqual(response.data['site_name'], 'Test Site') + self.assertEqual(response.data['river_name'], 'Test River') + self.assertEqual(response.data['description'], 'Test Description') + + # Ensure no images are attached to the site + site_id = response.data['gid'] + images_count = SiteImage.objects.filter(site_id=site_id).count() + self.assertEqual(images_count, 0) + + def test_list_site_filter(self): + Sites.objects.create( + user=self.user, + site_name='Cape Town', + the_geom=Point(0, 0) + ) + Sites.objects.create( + user=self.user, + site_name='Limpopo', + the_geom=Point(1, 1) + ) + client = APIClient() + client.force_authenticate(user=self.user) + url = reverse('sites-list-create') + response = client.get(f'{url}?site_name=mpo', format='json') + self.assertEquals(len(response.json()), 1) + self.assertEquals(response.json()[0]['site_name'], 'Limpopo') class SiteObservationsByLocationTestCase(APITestCase): - def setUp(self): - # Create a user - self.user = User.objects.create_user(username='testuser', password='testpassword') - - # Create a site with observations - self.site = Sites.objects.create( - site_name='Test Site', - river_name='Test River', - description='Test Description', - river_cat='rocky', - user=self.user, - the_geom='SRID=4326;POINT({} {})'.format(24.84165007535725, -30.47829136066817), - ) - Observations.objects.create( - flatworms=True, - worms=False, - leeches=False, - crabs_shrimps=True, - stoneflies=False, - minnow_mayflies=True, - other_mayflies=False, - damselflies=True, - dragonflies=False, - bugs_beetles=True, - caddisflies=False, - true_flies=False, - snails=False, - score="4.11", - time_stamp="2023-11-28T14:20:42.329190+02:00", - comment="test_observation", - obs_date="2023-11-28", - flag="clean", - water_clarity="2.0", - water_temp="1.2", - ph="1.0", - diss_oxygen="2.40", - diss_oxygen_unit="%DO", - elec_cond="2.50", - elec_cond_unit="mS/m", - site=self.site, - ) - - def test_get_site_observations_by_location(self): - url = reverse('site-observations', args=[self.site.the_geom.y, self.site.the_geom.x]) - response = self.client.get(url) - - self.assertEqual(response.status_code, status.HTTP_200_OK) - - expected_data = SitesWithObservationsSerializer(self.site).data - self.assertEqual(response.data, expected_data) - - def test_get_nonexistent_site_observations_by_location(self): - url = reverse('site-observations', args=[0.0, 0.0]) # Provide non-existent coordinates - response = self.client.get(url) - - self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) - self.assertEqual(response.data, []) + def setUp(self): + # Create a user + self.user = User.objects.create_user(username='testuser', password='testpassword') + + # Create a site with observations + self.site = Sites.objects.create( + site_name='Test Site', + river_name='Test River', + description='Test Description', + river_cat='rocky', + user=self.user, + the_geom='SRID=4326;POINT({} {})'.format(24.84165007535725, -30.47829136066817), + ) + Observations.objects.create( + flatworms=True, + worms=False, + leeches=False, + crabs_shrimps=True, + stoneflies=False, + minnow_mayflies=True, + other_mayflies=False, + damselflies=True, + dragonflies=False, + bugs_beetles=True, + caddisflies=False, + true_flies=False, + snails=False, + score="4.11", + time_stamp="2023-11-28T14:20:42.329190+02:00", + comment="test_observation", + obs_date="2023-11-28", + flag="clean", + water_clarity="2.0", + water_temp="1.2", + ph="1.0", + diss_oxygen="2.40", + diss_oxygen_unit="%DO", + elec_cond="2.50", + elec_cond_unit="mS/m", + site=self.site, + ) + + def test_get_site_observations_by_location(self): + url = reverse('site-observations', args=[self.site.the_geom.y, self.site.the_geom.x]) + response = self.client.get(url) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + + expected_data = SitesWithObservationsSerializer(self.site).data + self.assertEqual(response.data, expected_data) + + def test_get_nonexistent_site_observations_by_location(self): + url = reverse('site-observations', args=[0.0, 0.0]) # Provide non-existent coordinates + response = self.client.get(url) + + self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) + self.assertEqual(response.data, [])