From e359df5ec1225900bef059738467ba4154703299 Mon Sep 17 00:00:00 2001 From: jchemutt Date: Fri, 8 Nov 2024 19:08:44 +0300 Subject: [PATCH] Added Forages Models --- src/ormWP/models/forecast.py | 60 ++++++++++++++++++++++++ src/ormWP/models/trend.py | 59 +++++++++++++++++++++++ src/ormWP/models/woreda.py | 56 ++++++++++++++++++++++ src/tests/test_forecast.py | 91 ++++++++++++++++++++++++++++++++++++ src/tests/test_trend.py | 91 ++++++++++++++++++++++++++++++++++++ src/tests/test_woreda.py | 64 +++++++++++++++++++++++++ 6 files changed, 421 insertions(+) create mode 100644 src/ormWP/models/forecast.py create mode 100644 src/ormWP/models/trend.py create mode 100644 src/ormWP/models/woreda.py create mode 100644 src/tests/test_forecast.py create mode 100644 src/tests/test_trend.py create mode 100644 src/tests/test_woreda.py diff --git a/src/ormWP/models/forecast.py b/src/ormWP/models/forecast.py new file mode 100644 index 0000000..4f91858 --- /dev/null +++ b/src/ormWP/models/forecast.py @@ -0,0 +1,60 @@ +from mongoengine import Document, ReferenceField, FloatField, DateField, DictField +from datetime import datetime +from .woreda import Woreda + +class Forecast(Document): + """ + Represents a Forecast in the database. + + Attributes: + ---------- + woreda: ReferenceField + Reference to the associated Woreda. Required. + mean: float + Mean value for the forecast. Required. + date: date + Date of the forecast. Required and unique per woreda. + trace: dict + Contains metadata such as created time and updated time. + + Methods: + ------- + save() + Saves the Forecast object to the database, updating the 'updated_at' field in trace. + delete() + Deletes the Forecast object from the database. + """ + + meta = { + 'collection': 'forecasts', + 'indexes': [ + { + 'fields': ['woreda', 'date'], + 'unique': True + } + ] + } + + woreda = ReferenceField(Woreda, required=True, reverse_delete_rule=2) + mean = FloatField(required=True, precision=4) + date = DateField(required=True) + trace = DictField(default=lambda: { + 'created_at': datetime.now(), + 'updated_at': datetime.now() + }) + + def save(self, *args, **kwargs): + """Override save to update the 'updated_at' field.""" + if not self.trace: + self.trace = { + 'created_at': datetime.now(), + 'updated_at': datetime.now() + } + else: + self.trace['updated_at'] = datetime.now() + + + self._mark_as_changed('trace') + + return super().save(*args, **kwargs) + diff --git a/src/ormWP/models/trend.py b/src/ormWP/models/trend.py new file mode 100644 index 0000000..24e3b44 --- /dev/null +++ b/src/ormWP/models/trend.py @@ -0,0 +1,59 @@ +from mongoengine import Document, ReferenceField, FloatField, DateField, DictField +from datetime import datetime +from .woreda import Woreda + +class Trend(Document): + """ + Represents a Trend in the database. + + Attributes: + ---------- + woreda: ReferenceField + Reference to the associated Woreda. Required. + mean: float + Mean value for the trend. Required. + date: date + Date for the trend. Required and unique per woreda. + trace: dict + Contains metadata such as created time, updated time. + + Methods: + ------- + save() + Saves the Trend object to the database, updating the 'updated_at' field in trace. + delete() + Deletes the Trend object from the database. + """ + + meta = { + 'collection': 'trends', + 'indexes': [ + { + 'fields': ['woreda', 'date'], + 'unique': True + } + ] + } + + woreda = ReferenceField(Woreda, required=True, reverse_delete_rule=2) + mean = FloatField(required=True, precision=4) + date = DateField(required=True) + trace = DictField(default=lambda: { + 'created_at': datetime.now(), + 'updated_at': datetime.now() + }) + + def save(self, *args, **kwargs): + """Override save to update the 'updated_at' field.""" + if not self.trace: + self.trace = { + 'created_at': datetime.now(), + 'updated_at': datetime.now() + } + else: + self.trace['updated_at'] = datetime.now() + + + self._mark_as_changed('trace') + + return super().save(*args, **kwargs) diff --git a/src/ormWP/models/woreda.py b/src/ormWP/models/woreda.py new file mode 100644 index 0000000..9090947 --- /dev/null +++ b/src/ormWP/models/woreda.py @@ -0,0 +1,56 @@ +from mongoengine import Document, StringField, DictField +from datetime import datetime + +class Woreda(Document): + """ + Represents a Woreda in the database. + + Attributes: + ---------- + name: str + Name of the Woreda. Required. + ext_id: str + External identifier for the Woreda. Required and unique. + trace: dict + Contains metadata such as created time, updated time, and enabled status. + + Methods: + ------- + save() + Saves the Woreda object to the database, updating the 'updated_at' field in trace. + delete() + Deletes the Woreda object from the database. + """ + + meta = { + 'collection': 'woredas', + 'indexes': [ + { + 'fields': ['ext_id'], + 'unique': True + } + ] + } + + name = StringField(max_length=255, required=True) + ext_id = StringField(max_length=255, required=True, unique=True) + trace = DictField(default=lambda: { + 'created_at': datetime.now(), + 'updated_at': datetime.now(), + 'enabled': True + }) + + def save(self, *args, **kwargs): + """Override save to update the 'updated_at' field.""" + if not self.trace: + self.trace = { + 'created_at': datetime.now(), + 'updated_at': datetime.now(), + 'enabled': True + } + else: + self.trace['updated_at'] = datetime.now() + + self._mark_as_changed('trace') + + return super().save(*args, **kwargs) diff --git a/src/tests/test_forecast.py b/src/tests/test_forecast.py new file mode 100644 index 0000000..1d5f31d --- /dev/null +++ b/src/tests/test_forecast.py @@ -0,0 +1,91 @@ +import unittest +from datetime import datetime, date +from mongoengine import connect, disconnect +import sys +import os +dir_path = os.path.dirname(os.path.realpath(__file__)) +orm_dir_path = os.path.abspath(os.path.join(dir_path, '..')) +sys.path.append(orm_dir_path) +from ormWP.models.woreda import Woreda +from ormWP.models.forecast import Forecast + + +class TestForecast(unittest.TestCase): + def setUp(self): + disconnect() + connect('test_forecasts', host='mongomock://localhost') + + # Create a sample Woreda + self.woreda = Woreda( + name="Test Woreda", + ext_id="TEST001" + ) + self.woreda.save() + + # Create a sample Forecast + self.forecast = Forecast( + woreda=self.woreda, + mean=3.123, + date=date(2024, 11, 8) + ) + + def tearDown(self): + disconnect() + + def test_create_forecast(self): + self.forecast.save() + + # Assert that the Forecast has been saved and has an ID + self.assertIsNotNone(self.forecast.id) + + # Fetch the Forecast from the database + fetched_forecast = Forecast.objects(id=self.forecast.id).first() + + # Assert that the fetched Forecast matches the saved Forecast + self.assertEqual(fetched_forecast.mean, 3.123) + self.assertEqual(fetched_forecast.date, date(2024, 11, 8)) + self.assertEqual(fetched_forecast.trace['created_at'], fetched_forecast.trace['updated_at']) + + def test_update_forecast(self): + self.forecast.save() + + # Update the Forecast's mean value + self.forecast.mean = 2.456 + self.forecast.save() + + # Fetch the updated Forecast from the database + updated_forecast = Forecast.objects(id=self.forecast.id).first() + + # Assert that the Forecast's mean has been updated + self.assertEqual(updated_forecast.mean, 2.456) + + # Assert that the updated_at field has changed + self.assertNotEqual(updated_forecast.trace['updated_at'], updated_forecast.trace['created_at']) + self.assertTrue(updated_forecast.trace['updated_at'] > updated_forecast.trace['created_at']) + + def test_delete_forecast(self): + self.forecast.save() + + # Delete the Forecast + self.forecast.delete() + + # Assert that the Forecast no longer exists in the database + deleted_forecast = Forecast.objects(id=self.forecast.id).first() + self.assertIsNone(deleted_forecast) + + def test_unique_forecast_per_woreda_and_date(self): + self.forecast.save() + + # Attempt to create another Forecast with the same woreda and date + duplicate_forecast = Forecast( + woreda=self.woreda, + mean=1.789, + date=date(2024, 11, 8) + ) + + with self.assertRaises(Exception): + duplicate_forecast.save() + + +if __name__ == "__main__": + unittest.main() diff --git a/src/tests/test_trend.py b/src/tests/test_trend.py new file mode 100644 index 0000000..ec985e5 --- /dev/null +++ b/src/tests/test_trend.py @@ -0,0 +1,91 @@ +import unittest +from datetime import datetime, date, timezone +from mongoengine import connect, disconnect +import sys +import os +dir_path = os.path.dirname(os.path.realpath(__file__)) +orm_dir_path = os.path.abspath(os.path.join(dir_path, '..')) +sys.path.append(orm_dir_path) +from ormWP.models.woreda import Woreda +from ormWP.models.trend import Trend + + +class TestTrend(unittest.TestCase): + def setUp(self): + disconnect() + connect('test_trends', host='mongomock://localhost') + + # Create a sample Woreda + self.woreda = Woreda( + name="Test Woreda", + ext_id="TEST001" + ) + self.woreda.save() + + # Create a sample Trend + self.trend = Trend( + woreda=self.woreda, + mean=3.234, + date=date(2024, 11, 8) + ) + + def tearDown(self): + disconnect() + + def test_create_trend(self): + self.trend.save() + + # Assert that the Trend has been saved and has an ID + self.assertIsNotNone(self.trend.id) + + # Fetch the Trend from the database + fetched_trend = Trend.objects(id=self.trend.id).first() + + # Assert that the fetched Trend matches the saved Trend + self.assertEqual(fetched_trend.mean, 3.234) + self.assertEqual(fetched_trend.date, date(2024, 11, 8)) + self.assertEqual(fetched_trend.trace['created_at'], fetched_trend.trace['updated_at']) + + def test_update_trend(self): + self.trend.save() + + # Update the Trend's mean value + self.trend.mean = 2.123 + self.trend.save() + + # Fetch the updated Trend from the database + updated_trend = Trend.objects(id=self.trend.id).first() + + # Assert that the Trend's mean has been updated + self.assertEqual(updated_trend.mean, 2.123) + + # Assert that the updated_at field has changed + self.assertNotEqual(updated_trend.trace['updated_at'], updated_trend.trace['created_at']) + self.assertTrue(updated_trend.trace['updated_at'] > updated_trend.trace['created_at']) + + def test_delete_trend(self): + self.trend.save() + + # Delete the Trend + self.trend.delete() + + # Assert that the Trend no longer exists in the database + deleted_trend = Trend.objects(id=self.trend.id).first() + self.assertIsNone(deleted_trend) + + def test_unique_trend_per_woreda_and_date(self): + self.trend.save() + + # Attempt to create another Trend with the same woreda and date + duplicate_trend = Trend( + woreda=self.woreda, + mean=4.456, + date=date(2024, 11, 8) + ) + + with self.assertRaises(Exception): + duplicate_trend.save() + + +if __name__ == "__main__": + unittest.main() diff --git a/src/tests/test_woreda.py b/src/tests/test_woreda.py new file mode 100644 index 0000000..efd96c4 --- /dev/null +++ b/src/tests/test_woreda.py @@ -0,0 +1,64 @@ +import unittest +from mongoengine import connect, disconnect +import sys +import os +dir_path = os.path.dirname(os.path.realpath(__file__)) +orm_dir_path = os.path.abspath(os.path.join(dir_path, '..')) +sys.path.append(orm_dir_path) +from ormWP.models.woreda import Woreda + +class TestWoreda(unittest.TestCase): + def setUp(self): + disconnect() + connect('test_water_points', host='mongomock://localhost') + self.woreda = Woreda( + name="Test Woreda", + ext_id="TEST001" + ) + + def tearDown(self): + disconnect() + + def test_create_woreda(self): + # Save the Woreda to the database + self.woreda.save() + + # Assert that the Woreda has been saved and has an ID + self.assertIsNotNone(self.woreda.id) + + # Fetch the Woreda from the database + fetched_woreda = Woreda.objects(id=self.woreda.id).first() + + # Assert that the fetched Woreda matches the saved Woreda + self.assertEqual(fetched_woreda.name, "Test Woreda") + self.assertEqual(fetched_woreda.ext_id, "TEST001") + self.assertTrue(fetched_woreda.trace['enabled']) + + def test_update_woreda(self): + # Save the Woreda to the database + self.woreda.save() + + # Update the Woreda's name + self.woreda.name = "Updated Woreda" + self.woreda.save() + + # Fetch the updated Woreda from the database + updated_woreda = Woreda.objects(id=self.woreda.id).first() + + # Assert that the Woreda's name has been updated + self.assertEqual(updated_woreda.name, "Updated Woreda") + self.assertNotEqual(updated_woreda.trace['updated_at'], updated_woreda.trace['created_at']) + + def test_delete_woreda(self): + # Save the Woreda to the database + self.woreda.save() + + # Delete the Woreda + self.woreda.delete() + + # Assert that the Woreda no longer exists in the database + deleted_woreda = Woreda.objects(id=self.woreda.id).first() + self.assertIsNone(deleted_woreda) + +if __name__ == '__main__': + unittest.main()