Merge pull request #24 from CIAT-DAPA/develop

Develop

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)
+

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)

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)

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()

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()