Add EmissionFactorDS data model and update schema and fixtures

schemas/base_types.py

@@ -0,0 +1,6 @@
+from typing import Annotated
+
+from pydantic import Field
+
+
+NonEmptyString = Annotated[str, Field(min_length=1)]

schemas/carbon_footprint.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Annotated
 
 from pydantic import (
     AwareDatetime, BaseModel, ConfigDict, Field, PositiveFloat, confloat, constr,
     field_serializer
 )
 
+from schemas.base_types import NonEmptyString
+
 
 class CharacterizationFactors(str, Enum):
     """
@@ -133,6 +135,29 @@ class RegionOrSubregion(str, Enum):
     WESTERN_EUROPE = 'Western Europe'
 
 
+class EmissionFactorDS(BaseModel):
+    """Represents an EmissionFactorDS, referencing emission factor databases.
+
+    For further details, please refer to the Pathfinder Framework Section 4.1.3.2.
+
+    Attributes:
+        name (NonEmptyString): The name of the emission factor database. Must be non-empty.
+        version (NonEmptyString): The version of the emission factor database. Must be non-empty.
+
+    Each EmissionFactorDS MUST be encoded as a JSON object.
+    Example:
+        {
+          "name": "ecoinvent",
+          "version": "3.9.1"
+        }
+
+    For more information, please refer to the official documentation:
+        https://wbcsd.github.io/tr/2023/data-exchange-protocol-20231207/#dt-emissionfactords
+    """
+    name: NonEmptyString = Field(..., description="The non-empty name of the emission factor database.")
+    version: NonEmptyString = Field(..., description="The non-empty version of the emission factor database.")
+
+
 class CarbonFootprint(BaseModel):
     model_config = ConfigDict(
         use_enum_values=True
@@ -218,7 +243,7 @@ class CarbonFootprint(BaseModel):
         None, description='ISO 3166 Country Code if applicable.'
     )
     geographyRegionOrSubregion: Optional[RegionOrSubregion] = None
-    secondaryEmissionFactorSources: Optional[List[str]] = Field(
+    secondaryEmissionFactorSources: Optional[List[EmissionFactorDS]] = Field(
         None, description='List of emission factor sources for secondary data, if used.'
     )
     exemptedEmissionsPercent: confloat(ge=0.0, le=5.0) = Field(

tests/conftest.py

@@ -1,25 +1,26 @@
 import os
 import sys
+from datetime import datetime, timezone
 from typing import Any
 from typing import Generator
 
+import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
-import pytest
+from fastapi_pagination import add_pagination
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
-
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 # this is to include backend dir in sys.path so that we can import from db,main.py
 
+from apis.base import api_router
 from db.base import Base
 from db.session import get_db
 from db.repository.users import create_new_user
-from apis.base import api_router
 from schemas.user import UserCreate
-from core.pagination import PaginationMiddleware
-from fastapi_pagination import add_pagination
+
+
 
 
 def start_application():
@@ -99,3 +100,48 @@ def test_user(db_session: SessionTesting):
 about good quality fixture organisation in Pytest.
 """
 
+
+@pytest.fixture(scope="module")
+def valid_carbon_footprint_data():
+    data = {
+        "declaredUnit": "kilogram",
+        "unitaryProductAmount": 100,
+        "pCfExcludingBiogenic": 10,
+        "pCfIncludingBiogenic": 12,
+        "fossilGhgEmissions": 8,
+        "fossilCarbonContent": 5,
+        "biogenicCarbonContent": 4,
+        "dLucGhgEmissions": 2,
+        "landManagementGhgEmissions": 3,
+        "otherBiogenicGhgEmissions": 1,
+        "iLucGhgEmissions": 2,
+        "biogenicCarbonWithdrawal": -1,
+        "aircraftGhgEmissions": 0.5,
+        "characterizationFactors": "AR6",
+        "crossSectoralStandardsUsed": ["PAS 2050"],
+        "productOrSectorSpecificRules": ["CFS Guidance for XYZ Sector"],
+        "biogenicAccountingMethodology": "PEF",
+        "boundaryProcessesDescription": "Description of boundary processes",
+        "referencePeriodStart": datetime(2023, 1, 1, tzinfo=timezone.utc).isoformat(),
+        "referencePeriodEnd": datetime(2023, 12, 31, tzinfo=timezone.utc).isoformat(),
+        "geographyCountrySubdivision": "AU",
+        "geographyCountry": "AU",
+        "geographyRegionOrSubregion": "Australia and New Zealand",
+        "secondaryEmissionFactorSources": [
+            {
+                "name": "ecoinvent",
+                "version": "3.9.1"
+            }
+        ],
+        "exemptedEmissionsPercent": 2.5,
+        "exemptedEmissionsDescription": "Description of exempted emissions",
+        "packagingEmissionsIncluded": True,
+        "packagingGhgEmissions": 0.5,
+        "allocationRulesDescription": "Description of allocation rules",
+        "uncertaintyAssessmentDescription": "Description of uncertainty assessment",
+        "primaryDataShare": 50,
+        "dqi": {"key1": "value1", "key2": "value2"},
+        "assurance": {"key1": "value1", "key2": "value2"}
+    }
+
+    return data

tests/test_repository/test_product_footprints_repository.py

@@ -1,58 +1,15 @@
 from datetime import datetime, timezone
-from unittest import mock
 
 import pytest
-import sqlalchemy
-import pydantic
 
-#from db.models.product_footprint import ProductFootprint
-from db.repository.product_footprints import create_new_product_footprint, retrieve_product_footprint, \
-    list_product_footprints, count_product_footprints
+from db.repository.product_footprints import (
+    create_new_product_footprint, retrieve_product_footprint, list_product_footprints,
+    count_product_footprints
+)
 from schemas.product_footprint import ProductFootprint
 from schemas.carbon_footprint import CarbonFootprint
 
 
-@pytest.fixture(scope="module")
-def valid_carbon_footprint_data():
-    data = {
-        "declaredUnit": "kilogram",
-        "unitaryProductAmount": 100,
-        "pCfExcludingBiogenic": 10,
-        "pCfIncludingBiogenic": 12,
-        "fossilGhgEmissions": 8,
-        "fossilCarbonContent": 5,
-        "biogenicCarbonContent": 4,
-        "dLucGhgEmissions": 2,
-        "landManagementGhgEmissions": 3,
-        "otherBiogenicGhgEmissions": 1,
-        "iLucGhgEmissions": 2,
-        "biogenicCarbonWithdrawal": -1,
-        "aircraftGhgEmissions": 0.5,
-        "characterizationFactors": "AR6",
-        "crossSectoralStandardsUsed": ["PAS 2050"],
-        "productOrSectorSpecificRules": ["CFS Guidance for XYZ Sector"],
-        "biogenicAccountingMethodology": "PEF",
-        "boundaryProcessesDescription": "Description of boundary processes",
-        "referencePeriodStart": datetime(2023, 1, 1, tzinfo=timezone.utc).isoformat(),
-        "referencePeriodEnd": datetime(2023, 12, 31, tzinfo=timezone.utc).isoformat(),
-        "geographyCountrySubdivision": "AU",
-        "geographyCountry": "AU",
-        "geographyRegionOrSubregion": "Australia and New Zealand",
-        "secondaryEmissionFactorSources": ["Source 1", "Source 2"],
-        "exemptedEmissionsPercent": 2.5,
-        "exemptedEmissionsDescription": "Description of exempted emissions",
-        "packagingEmissionsIncluded": True,
-        "packagingGhgEmissions": 0.5,
-        "allocationRulesDescription": "Description of allocation rules",
-        "uncertaintyAssessmentDescription": "Description of uncertainty assessment",
-        "primaryDataShare": 50,
-        "dqi": {"key1": "value1", "key2": "value2"},
-        "assurance": {"key1": "value1", "key2": "value2"}
-    }
-
-    return data
-
-
 @pytest.fixture(scope="module")
 def valid_product_footprint_data(valid_carbon_footprint_data):
     data = {

tests/test_routes/test_product_footprints_route.py

@@ -1,6 +1,3 @@
-import json
-import uuid
-
 import pytest
 
 
@@ -58,7 +55,12 @@ def valid_json_product_footprint():
             "geographyCountrySubdivision": "string",
             "geographyCountry": "AU",
             "geographyRegionOrSubregion": "Australia and New Zealand",
-            "secondaryEmissionFactorSources": ["Ecoinvent"],
+            "secondaryEmissionFactorSources": [
+                {
+                    "name": "ecoinvent",
+                    "version": "3.9.1"
+                }
+            ],
             "exemptedEmissionsPercent": 0,
             "exemptedEmissionsDescription": "string",
             "packagingEmissionsIncluded": True,

tests/test_schemas/test_base_types.py

@@ -0,0 +1,23 @@
+import pytest
+from pydantic import BaseModel, ValidationError
+
+
+from schemas.base_types import NonEmptyString
+
+
+def test_non_empty_string_valid():
+    class Model(BaseModel):
+        value: NonEmptyString
+
+    Model(value="hello")
+
+
+def test_non_empty_string_empty():
+    class Model(BaseModel):
+        value: NonEmptyString
+
+    with pytest.raises(ValidationError) as excinfo:
+        Model(value="")
+
+    assert excinfo.value.errors()[0]["loc"] == ("value",)
+    assert excinfo.value.errors()[0]["msg"] == "String should have at least 1 character"

tests/test_schemas/test_carbon_footprint_schema.py

@@ -9,47 +9,6 @@
 )
 
 
-@pytest.fixture(scope="function")
-def valid_carbon_footprint_data():
-    data = {
-        "declaredUnit": "kilogram",
-        "unitaryProductAmount": 100,
-        "pCfExcludingBiogenic": 10,
-        "pCfIncludingBiogenic": 12,
-        "fossilGhgEmissions": 8,
-        "fossilCarbonContent": 5,
-        "biogenicCarbonContent": 4,
-        "dLucGhgEmissions": 2,
-        "landManagementGhgEmissions": 3,
-        "otherBiogenicGhgEmissions": 1,
-        "iLucGhgEmissions": 2,
-        "biogenicCarbonWithdrawal": -1,
-        "aircraftGhgEmissions": 0.5,
-        "characterizationFactors": "AR6",
-        "crossSectoralStandardsUsed": ["PAS 2050"],
-        "productOrSectorSpecificRules": ["CFS Guidance for XYZ Sector"],
-        "biogenicAccountingMethodology": "PEF",
-        "boundaryProcessesDescription": "Description of boundary processes",
-        "referencePeriodStart": datetime(2023, 1, 1, tzinfo=timezone.utc).isoformat(),
-        "referencePeriodEnd": datetime(2023, 12, 31, tzinfo=timezone.utc).isoformat(),
-        "geographyCountrySubdivision": "AU",
-        "geographyCountry": "AU",
-        "geographyRegionOrSubregion": "Australia and New Zealand",
-        "secondaryEmissionFactorSources": ["Source 1", "Source 2"],
-        "exemptedEmissionsPercent": 2.5,
-        "exemptedEmissionsDescription": "Description of exempted emissions",
-        "packagingEmissionsIncluded": True,
-        "packagingGhgEmissions": 0.5,
-        "allocationRulesDescription": "Description of allocation rules",
-        "uncertaintyAssessmentDescription": "Description of uncertainty assessment",
-        "primaryDataShare": 50,
-        "dqi": {"key1": "value1", "key2": "value2"},
-        "assurance": {"key1": "value1", "key2": "value2"}
-    }
-
-    return data
-
-
 def test_carbon_footprint_schema_validation(valid_carbon_footprint_data):
     """Test successful validation for valid carbon footprint data."""
     footprint = CarbonFootprint(**valid_carbon_footprint_data)
@@ -79,7 +38,12 @@ def test_carbon_footprint_schema_validation(valid_carbon_footprint_data):
     assert json_footprint_data["geographyCountrySubdivision"] == "AU"
     assert json_footprint_data["geographyCountry"] == "AU"
     assert json_footprint_data["geographyRegionOrSubregion"] == "Australia and New Zealand"
-    assert json_footprint_data["secondaryEmissionFactorSources"] == ["Source 1", "Source 2"]
+    assert json_footprint_data["secondaryEmissionFactorSources"] == [
+        {
+            "name": "ecoinvent",
+            "version": "3.9.1"
+        }
+    ]
     assert json_footprint_data["exemptedEmissionsPercent"] == 2.5
     assert json_footprint_data["exemptedEmissionsDescription"] == "Description of exempted emissions"
     assert json_footprint_data["packagingEmissionsIncluded"] == True

tests/test_schemas/test_emission_factor_ds_schema.py

@@ -0,0 +1,53 @@
+import pytest
+from pydantic import ValidationError
+
+from schemas.carbon_footprint import EmissionFactorDS
+
+
+def test_emission_factor_valid():
+    data = {
+        "name": "ecoinvent",
+        "version": "2.3.0"
+    }
+    emission_factor_ds = EmissionFactorDS(**data)
+
+    assert emission_factor_ds.name == "ecoinvent"
+    assert emission_factor_ds.version == "2.3.0"
+
+
+def test_emission_factor_missing_name():
+    data = {
+        "version": "1.0.0"
+    }
+    with pytest.raises(ValidationError):
+        EmissionFactorDS(**data)
+
+
+def test_emission_factor_empty_version():
+    data = {
+        "name": "ecoinvent",
+        "version": ""
+    }
+    with pytest.raises(ValidationError):
+        EmissionFactorDS(**data)
+
+
+def test_emission_factor_ds_json_representation():
+    data = {
+        "name": "ecoinvent",
+        "version": "3.9.1"
+    }
+    emission_factor_ds = EmissionFactorDS(**data)
+
+    expected_json = '{"name":"ecoinvent","version":"3.9.1"}'
+    assert emission_factor_ds.model_dump_json() == expected_json
+
+
+# def test_emission_factor_ds_json_representation():
+#     emission_factor_ds = EmissionFactorDS(name="ecoinvent", version="3.9.1")
+#     expected_json = '{"name":"ecoinvent","version":"3.9.1"}'
+#
+#     assert emission_factor_ds.model_dump_json() == expected_json
+#
+#     # Additionally, ensure the dict representation is as expected using .dict()
+#     assert emission_factor_ds.model_dump() == {"name": "ecoinvent", "version": "3.9.1"}