From 7c539c81584c07df7fc64a074b7e648c9eed08f5 Mon Sep 17 00:00:00 2001 From: "K. Allagbe" Date: Mon, 14 Oct 2024 02:26:14 -0400 Subject: [PATCH] issue #158: province functions module --- fertiscan/db/queries/organization/__init__.py | 93 ------------- fertiscan/db/queries/province/__init__.py | 127 ++++++++++++++++++ tests/fertiscan/db/queries/test_fertilizer.py | 14 +- tests/fertiscan/db/queries/test_location.py | 19 ++- .../fertiscan/db/queries/test_organization.py | 48 ++----- tests/fertiscan/db/queries/test_province.py | 126 +++++++++++++++++ .../db/queries/test_update_guaranteed.py | 8 +- .../db/queries/test_update_metrics.py | 8 +- .../db/queries/test_update_sub_labels.py | 8 +- 9 files changed, 291 insertions(+), 160 deletions(-) create mode 100644 fertiscan/db/queries/province/__init__.py create mode 100644 tests/fertiscan/db/queries/test_province.py diff --git a/fertiscan/db/queries/organization/__init__.py b/fertiscan/db/queries/organization/__init__.py index f742fb64..f9f0e076 100644 --- a/fertiscan/db/queries/organization/__init__.py +++ b/fertiscan/db/queries/organization/__init__.py @@ -17,14 +17,6 @@ class OrganizationUpdateError(Exception): pass -class ProvinceCreationError(Exception): - pass - - -class ProvinceNotFoundError(Exception): - pass - - def new_organization(cursor, information_id, location_id=None): """ This function create a new organization in the database. @@ -386,88 +378,3 @@ def get_full_organization(cursor, org_id): return cursor.fetchone() except Exception as e: raise Exception("Datastore organization unhandeled error" + e.__str__()) - - -def new_province(cursor, name): - """ - This function create a new province in the database. - - Parameters: - - cursor (cursor): The cursor of the database. - - name (str): The name of the province. - - Returns: - - str: The UUID of the province - """ - try: - query = """ - INSERT INTO - province ( - name - ) - VALUES - (%s) - RETURNING - id - """ - cursor.execute(query, (name,)) - return cursor.fetchone()[0] - except Exception as e: - raise ProvinceCreationError("Datastore province unhandeled error" + e.__str__()) - - -def get_province(cursor, province_id): - """ - This function get a province from the database. - - Parameters: - - cursor (cursor): The cursor of the database. - - province_id (int): The UUID of the province. - - Returns: - - dict: The province - """ - try: - query = """ - SELECT - name - FROM - province - WHERE - id = %s - """ - cursor.execute(query, (province_id,)) - res = cursor.fetchone() - if res is None: - raise ProvinceNotFoundError - return res - except ProvinceNotFoundError: - raise ProvinceNotFoundError( - "province not found with province_id: " + str(province_id) - ) - except Exception as e: - raise Exception("Datastore organization unhandeled error" + e.__str__()) - - -def get_all_province(cursor): - """ - This function get all province from the database. - - Parameters: - - cursor (cursor): The cursor of the database. - - Returns: - - dict: The province - """ - try: - query = """ - SELECT - id, - name - FROM - province - """ - cursor.execute(query) - return cursor.fetchall() - except Exception as e: - raise Exception("Datastore organization unhandeled error" + e.__str__()) diff --git a/fertiscan/db/queries/province/__init__.py b/fertiscan/db/queries/province/__init__.py new file mode 100644 index 00000000..70c46429 --- /dev/null +++ b/fertiscan/db/queries/province/__init__.py @@ -0,0 +1,127 @@ +from psycopg import Cursor +from psycopg.rows import dict_row +from psycopg.sql import SQL + + +def create_province(cursor: Cursor, name: str) -> dict | None: + """ + Inserts a new province record into the database. + + Args: + cursor: Database cursor object. + name: Name of the province. + + Returns: + The inserted province record as a dictionary, or None if failed. + """ + query = SQL(""" + INSERT INTO province (name) + VALUES (%s) + RETURNING *; + """) + with cursor.connection.cursor(row_factory=dict_row) as new_cur: + new_cur.execute(query, (name,)) + return new_cur.fetchone() + + +def read_province(cursor: Cursor, province_id: int) -> dict | None: + """ + Retrieves a province record by ID. + + Args: + cursor: Database cursor object. + province_id: ID of the province. + + Returns: + The province record as a dictionary, or None if not found. + """ + query = SQL("SELECT * FROM province WHERE id = %s;") + with cursor.connection.cursor(row_factory=dict_row) as new_cur: + new_cur.execute(query, (province_id,)) + return new_cur.fetchone() + + +def read_all_provinces(cursor: Cursor) -> list[dict]: + """ + Retrieves all province records from the database. + + Args: + cursor: Database cursor object. + + Returns: + A list of all province records as dictionaries. + """ + query = SQL("SELECT * FROM province;") + with cursor.connection.cursor(row_factory=dict_row) as new_cur: + new_cur.execute(query) + return new_cur.fetchall() + + +def update_province(cursor: Cursor, province_id: int, name: str) -> dict | None: + """ + Updates an existing province record by ID. + + Args: + cursor: Database cursor object. + province_id: ID of the province. + name: New name of the province. + + Returns: + The updated province record as a dictionary, or None if not found. + """ + query = SQL(""" + UPDATE province + SET name = %s + WHERE id = %s + RETURNING *; + """) + with cursor.connection.cursor(row_factory=dict_row) as new_cur: + new_cur.execute(query, (name, province_id)) + return new_cur.fetchone() + + +def delete_province(cursor: Cursor, province_id: int) -> dict | None: + """ + Deletes a province record by ID. + + Args: + cursor: Database cursor object. + province_id: ID of the province. + + Returns: + The deleted province record as a dictionary, or None if not found. + """ + query = SQL(""" + DELETE FROM province + WHERE id = %s + RETURNING *; + """) + with cursor.connection.cursor(row_factory=dict_row) as new_cur: + new_cur.execute(query, (province_id,)) + return new_cur.fetchone() + + +def query_provinces(cursor: Cursor, name: str | None = None) -> list[dict]: + """ + Queries provinces based on optional filter criteria. + + Args: + cursor: Database cursor object. + name: Optional name to filter provinces. + + Returns: + A list of province records matching the filter criteria as dictionaries. + """ + conditions = [] + parameters = [] + + if name is not None: + conditions.append("name = %s") + parameters.append(name) + + where_clause = " WHERE " + " AND ".join(conditions) if conditions else "" + query = SQL(f"SELECT * FROM province{where_clause};") + + with cursor.connection.cursor(row_factory=dict_row) as new_cur: + new_cur.execute(query, parameters) + return new_cur.fetchall() diff --git a/tests/fertiscan/db/queries/test_fertilizer.py b/tests/fertiscan/db/queries/test_fertilizer.py index 9b7b0923..2d6fdbf3 100644 --- a/tests/fertiscan/db/queries/test_fertilizer.py +++ b/tests/fertiscan/db/queries/test_fertilizer.py @@ -7,7 +7,7 @@ from psycopg import Connection, connect from datastore.db.queries.user import register_user -from fertiscan.db.models import Fertilizer, Location, Region +from fertiscan.db.models import Fertilizer, Location, Province, Region from fertiscan.db.queries.fertilizer import ( create_fertilizer, delete_fertilizer, @@ -19,11 +19,8 @@ ) from fertiscan.db.queries.inspection import new_inspection from fertiscan.db.queries.location import create_location -from fertiscan.db.queries.organization import ( - new_organization, - new_organization_info, - new_province, -) +from fertiscan.db.queries.organization import new_organization, new_organization_info +from fertiscan.db.queries.province import create_province from fertiscan.db.queries.region import create_region load_dotenv() @@ -51,8 +48,9 @@ def setUp(self): self.inspector_id = register_user( self.cursor, f"{uuid.uuid4().hex}@example.com" ) - self.province_id = new_province(self.cursor, "a-test-province") - self.region = create_region(self.cursor, "test-region", self.province_id) + self.province = create_province(self.cursor, "a-test-province") + self.province = Province.model_validate(self.province) + self.region = create_region(self.cursor, "test-region", self.province.id) self.region = Region.model_validate(self.region) self.location = create_location( self.cursor, "test-location", "test-address", self.region.id diff --git a/tests/fertiscan/db/queries/test_location.py b/tests/fertiscan/db/queries/test_location.py index dfab90a3..4cdb3fd5 100644 --- a/tests/fertiscan/db/queries/test_location.py +++ b/tests/fertiscan/db/queries/test_location.py @@ -6,7 +6,7 @@ from psycopg import Connection, connect from datastore.db.queries.user import register_user -from fertiscan.db.models import FullLocation, Location, Region +from fertiscan.db.models import FullLocation, Location, Province, Region from fertiscan.db.queries.location import ( create_location, delete_location, @@ -16,11 +16,8 @@ update_location, upsert_location, ) -from fertiscan.db.queries.organization import ( - new_organization, - new_organization_info, - new_province, -) +from fertiscan.db.queries.organization import new_organization, new_organization_info +from fertiscan.db.queries.province import create_province from fertiscan.db.queries.region import create_region load_dotenv() @@ -45,8 +42,9 @@ def setUp(self): self.cursor = self.conn.cursor() # Create necessary records for testing - self.province_id = new_province(self.cursor, uuid.uuid4().hex) - self.region = create_region(self.cursor, "Test Region", self.province_id) + self.province = create_province(self.cursor, uuid.uuid4().hex) + self.province = Province.model_validate(self.province) + self.region = create_region(self.cursor, "Test Region", self.province.id) self.region = Region.model_validate(self.region) self.inspector_id = register_user(self.cursor, "inspector@example.com") @@ -270,8 +268,9 @@ def test_get_full_location_with_region_and_province(self): # Create province and region province_name = uuid.uuid4().hex region_name = "Test Region" - province_id = new_province(self.cursor, province_name) - region = create_region(self.cursor, region_name, province_id) + province = create_province(self.cursor, province_name) + province = Province.model_validate(province) + region = create_region(self.cursor, region_name, province.id) region = Region.model_validate(region) # Create a location with the new region diff --git a/tests/fertiscan/db/queries/test_organization.py b/tests/fertiscan/db/queries/test_organization.py index fcf7c31c..22d73950 100644 --- a/tests/fertiscan/db/queries/test_organization.py +++ b/tests/fertiscan/db/queries/test_organization.py @@ -9,9 +9,10 @@ import datastore.db as db from datastore.db.metadata import validator -from fertiscan.db.models import Location, Region +from fertiscan.db.models import Location, Province, Region from fertiscan.db.queries import label, organization from fertiscan.db.queries.location import create_location, query_locations +from fertiscan.db.queries.province import create_province from fertiscan.db.queries.region import create_region DB_CONNECTION_STRING = os.environ.get("FERTISCAN_DB_URL") @@ -23,40 +24,6 @@ raise ValueError("FERTISCAN_SCHEMA_TESTING is not set") -class test_province(unittest.TestCase): - def setUp(self): - self.con = db.connect_db(DB_CONNECTION_STRING, DB_SCHEMA) - self.cursor = self.con.cursor() - db.create_search_path(self.con, self.cursor, DB_SCHEMA) - - self.name = "test-province" - - def tearDown(self): - self.con.rollback() - db.end_query(self.con, self.cursor) - - def test_new_province(self): - province_id = organization.new_province(self.cursor, self.name) - self.assertIsInstance(province_id, int) - - def test_get_province(self): - province_id = organization.new_province(self.cursor, self.name) - province_data = organization.get_province(self.cursor, province_id) - self.assertEqual(province_data[0], self.name) - - def test_get_province_not_found(self): - with self.assertRaises(organization.ProvinceNotFoundError): - organization.get_province(self.cursor, 0) - - def test_get_all_province(self): - province_id = organization.new_province(self.cursor, self.name) - province_2_id = organization.new_province(self.cursor, "test-province-2") - province_data = organization.get_all_province(self.cursor) - self.assertEqual(len(province_data), 2) - self.assertEqual(province_data[0][0], province_id) - self.assertEqual(province_data[1][0], province_2_id) - - class test_organization_information(unittest.TestCase): def setUp(self): self.con = db.connect_db(DB_CONNECTION_STRING, DB_SCHEMA) @@ -69,9 +36,9 @@ def setUp(self): self.phone = "123456789" self.location_name = "test-location" self.location_address = "test-address" - self.province_id = organization.new_province(self.cursor, self.province_name) - - self.region = create_region(self.cursor, self.region_name, self.province_id) + self.province = create_province(self.cursor, self.province_name) + self.province = Province.model_validate(self.province) + self.region = create_region(self.cursor, self.region_name, self.province.id) self.region = Region.model_validate(self.region) self.location = create_location( @@ -235,8 +202,9 @@ def setUp(self): self.phone = "123456789" self.location_name = "test-location" self.location_address = "test-address" - self.province_id = organization.new_province(self.cursor, self.province_name) - self.region = create_region(self.cursor, self.region_name, self.province_id) + self.province = create_province(self.cursor, self.province_name) + self.province = Province.model_validate(self.province) + self.region = create_region(self.cursor, self.region_name, self.province.id) self.region = Region.model_validate(self.region) self.location = create_location( self.cursor, self.location_name, self.location_address, self.region.id diff --git a/tests/fertiscan/db/queries/test_province.py b/tests/fertiscan/db/queries/test_province.py new file mode 100644 index 00000000..4c9380ef --- /dev/null +++ b/tests/fertiscan/db/queries/test_province.py @@ -0,0 +1,126 @@ +import os +import unittest + +from dotenv import load_dotenv +from psycopg import Connection, connect + +from fertiscan.db.models import Province +from fertiscan.db.queries.province import ( + create_province, + delete_province, + query_provinces, + read_all_provinces, + read_province, + update_province, +) + +load_dotenv() + +TEST_DB_CONNECTION_STRING = os.environ["FERTISCAN_DB_URL"] +TEST_DB_SCHEMA = os.environ["FERTISCAN_SCHEMA_TESTING"] + + +class TestProvinceFunctions(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.conn: Connection = connect( + TEST_DB_CONNECTION_STRING, options=f"-c search_path={TEST_DB_SCHEMA},public" + ) + cls.conn.autocommit = False + + @classmethod + def tearDownClass(cls): + cls.conn.close() + + def setUp(self): + self.cursor = self.conn.cursor() + + def tearDown(self): + self.conn.rollback() + self.cursor.close() + + def test_create_province(self): + name = "Test Province A" + created_province = create_province(self.cursor, name) + validated_province = Province.model_validate(created_province) + + self.assertEqual(validated_province.name, name) + + def test_read_province(self): + name = "Test Province B" + created_province = create_province(self.cursor, name) + validated_province = Province.model_validate(created_province) + + fetched_province = read_province(self.cursor, validated_province.id) + fetched_province = Province.model_validate(fetched_province) + + self.assertEqual(fetched_province, validated_province) + + def test_read_all_provinces(self): + initial_provinces = read_all_provinces(self.cursor) + + province_a = create_province(self.cursor, "Test Province C") + province_b = create_province(self.cursor, "Test Province D") + + all_provinces = read_all_provinces(self.cursor) + all_provinces = [Province.model_validate(p) for p in all_provinces] + + self.assertGreaterEqual(len(all_provinces), len(initial_provinces) + 2) + self.assertIn(Province.model_validate(province_a), all_provinces) + self.assertIn(Province.model_validate(province_b), all_provinces) + + def test_update_province(self): + province = create_province(self.cursor, "Test Province E") + validated_province = Province.model_validate(province) + + new_name = "Updated Province E" + updated_province = update_province(self.cursor, validated_province.id, new_name) + validated_updated = Province.model_validate(updated_province) + + self.assertEqual(validated_updated.name, new_name) + + fetched_province = read_province(self.cursor, validated_province.id) + fetched_province = Province.model_validate(fetched_province) + + self.assertEqual(fetched_province, validated_updated) + + def test_delete_province(self): + province = create_province(self.cursor, "Test Province F") + validated_province = Province.model_validate(province) + + deleted_province = delete_province(self.cursor, validated_province.id) + deleted_province = Province.model_validate(deleted_province) + + self.assertEqual(validated_province, deleted_province) + + fetched_province = read_province(self.cursor, validated_province.id) + self.assertIsNone(fetched_province) + + def test_query_province_by_name(self): + name = "Test Province G" + create_province(self.cursor, name) + + result = query_provinces(self.cursor, name=name) + provinces = [Province.model_validate(p) for p in result] + + self.assertTrue(len(provinces) > 0) + self.assertEqual(provinces[0].name, name) + + def test_query_province_without_name(self): + """Test querying provinces without any filter.""" + # Create two provinces for testing + province_a = create_province(self.cursor, "Test Province H") + province_b = create_province(self.cursor, "Test Province I") + + # Query without any filter + result = query_provinces(self.cursor) + provinces = [Province.model_validate(p) for p in result] + + # Ensure the newly created provinces are present in the results + self.assertIn(Province.model_validate(province_a), provinces) + self.assertIn(Province.model_validate(province_b), provinces) + self.assertTrue(len(provinces) > 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/fertiscan/db/queries/test_update_guaranteed.py b/tests/fertiscan/db/queries/test_update_guaranteed.py index 0166b2c3..a78d120f 100644 --- a/tests/fertiscan/db/queries/test_update_guaranteed.py +++ b/tests/fertiscan/db/queries/test_update_guaranteed.py @@ -5,9 +5,10 @@ import psycopg from dotenv import load_dotenv -from fertiscan.db.models import GuaranteedAnalysis, Location, Region +from fertiscan.db.models import GuaranteedAnalysis, Location, Province, Region from fertiscan.db.queries import label, nutrients, organization from fertiscan.db.queries.location import create_location +from fertiscan.db.queries.province import create_province from fertiscan.db.queries.region import create_region load_dotenv() @@ -71,8 +72,9 @@ def setUp(self): self.phone = "123456789" self.location_name = "test-location" self.location_address = "test-address" - self.province_id = organization.new_province(self.cursor, self.province_name) - self.region = create_region(self.cursor, self.region_name, self.province_id) + self.province = create_province(self.cursor, self.province_name) + self.province = Province.model_validate(self.province) + self.region = create_region(self.cursor, self.region_name, self.province.id) self.region = Region.model_validate(self.region) self.location = create_location( self.cursor, self.location_name, self.location_address, self.region.id diff --git a/tests/fertiscan/db/queries/test_update_metrics.py b/tests/fertiscan/db/queries/test_update_metrics.py index f9a0bf71..2f80e736 100644 --- a/tests/fertiscan/db/queries/test_update_metrics.py +++ b/tests/fertiscan/db/queries/test_update_metrics.py @@ -5,9 +5,10 @@ from dotenv import load_dotenv import fertiscan.db.queries.label as label -from fertiscan.db.models import Location, Metric, Metrics, Region +from fertiscan.db.models import Location, Metric, Metrics, Province, Region from fertiscan.db.queries import metric, organization from fertiscan.db.queries.location import create_location +from fertiscan.db.queries.province import create_province from fertiscan.db.queries.region import create_region load_dotenv() @@ -52,8 +53,9 @@ def setUp(self): self.phone = "123456789" self.location_name = "test-location" self.location_address = "test-address" - self.province_id = organization.new_province(self.cursor, self.province_name) - self.region = create_region(self.cursor, self.region_name, self.province_id) + self.province = create_province(self.cursor, self.province_name) + self.province = Province.model_validate(self.province) + self.region = create_region(self.cursor, self.region_name, self.province.id) self.region = Region.model_validate(self.region) self.location = create_location( self.cursor, self.location_name, self.location_address, self.region.id diff --git a/tests/fertiscan/db/queries/test_update_sub_labels.py b/tests/fertiscan/db/queries/test_update_sub_labels.py index 1abd9712..b2c524d4 100644 --- a/tests/fertiscan/db/queries/test_update_sub_labels.py +++ b/tests/fertiscan/db/queries/test_update_sub_labels.py @@ -7,9 +7,10 @@ import fertiscan.db.queries.label as label import fertiscan.db.queries.sub_label as sub_label -from fertiscan.db.models import Inspection, Location, Region, SubLabel +from fertiscan.db.models import Inspection, Location, Province, Region, SubLabel from fertiscan.db.queries import organization from fertiscan.db.queries.location import create_location +from fertiscan.db.queries.province import create_province from fertiscan.db.queries.region import create_region load_dotenv() @@ -97,8 +98,9 @@ def setUp(self): self.phone = "123456789" self.location_name = "test-location" self.location_address = "test-address" - self.province_id = organization.new_province(self.cursor, self.province_name) - self.region = create_region(self.cursor, self.region_name, self.province_id) + self.province = create_province(self.cursor, self.province_name) + self.province = Province.model_validate(self.province) + self.region = create_region(self.cursor, self.region_name, self.province.id) self.region = Region.model_validate(self.region) self.location = create_location( self.cursor, self.location_name, self.location_address, self.region.id