Update routes and tests (#91)

* Updated deprecated methods

* Get all respondent ids on a specific data

* Update routes and tests

Created a new route for get respondent id feature and updated tests for the same

API/fwapp.py

@@ -266,6 +266,32 @@ def scoped_checks(user_creds):
     create_new_village(dbname, user_creds, response_result)
     return response_result
 
+@app.get('/api/get_respid_list', summary="Get the list of users", dependencies=[Depends(JWTBearer())], tags=["Resource Server"])
+async def get_respid_list(date:str, village_name:str=None, user_credentials:str=Depends(JWTBearer())):
+    response_result={
+        "status":"not_allowed",
+        "message":["Not authenticated"],
+        "data":{}
+    }
+
+    user_creds=get_current_user_credentials(user_credentials)
+
+    @scopes.init_checks(authorized_roles=['admin','GOVTOff'],response_result=response_result,village_name=village_name)
+    def scoped_checks(user_creds:UserOut):
+        if user_creds.role == 'admin':
+            response_data = get_resp_id_on_date(user_creds.village_name,'meta',date,response_result)
+        else:
+            response_data = get_resp_id_on_date(village_name,'meta',date,response_result)
+        return response_data
+
+    response_data = scoped_checks(user_creds)
+
+    response_result['data']['user_ids'] = response_data
+    response_result['status'] = 'success'
+    response_result['message'] = ['authorized']
+
+    return response_result
+
 # @app.get('/auth/me', summary='Get details of currently logged in user', response_model=UserOut, tags=["SessionInfo"])
 # async def get_me(user: str = Depends(JWTBearer())):
 #     data = get_current_user_credentials(user)

API/services/db/DBManipulation.py

@@ -87,67 +87,67 @@ def commit_to_db(response_result: dict, form_data: FormData, user_AADHAR: str)->
     fid = DBQueries.fetch_last(db, collection_names['meta'])['_id']
 
     # respondent's profile
-    data = form_data.respondent_prof.dict()
+    data = form_data.respondent_prof.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['rpf'], data)
 
     # gen_ho_data
-    data = form_data.gen_ho_info.dict()
+    data = form_data.gen_ho_info.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['ghi'], data)
 
     # fam_info
     data = form_data.fam_info
-    data = [fam_mem_info.dict() for fam_mem_info in data]
+    data = [fam_mem_info.model_dump() for fam_mem_info in data]
     [indiv_info.update({"__id": fid}) for indiv_info in data]
     DBQueries.insert_to_database(db, collection_names['fi'], data)
 
     # migration info
-    data = form_data.mig_status.dict()
+    data = form_data.mig_status.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['ms'], data)
 
     # gov schemes
-    data = form_data.govt_schemes.dict()
+    data = form_data.govt_schemes.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['gs'], data)
 
     # water source
-    data = form_data.water_source.dict()
+    data = form_data.water_source.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['ws'], data)
 
     # soucre of E
-    data = form_data.source_of_energy.dict()
+    data = form_data.source_of_energy.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['soe'], data)
 
     # Land holding info
-    data = form_data.land_holding_info.dict()
+    data = form_data.land_holding_info.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['lhi'], data)
 
     # agri inputs
-    data = form_data.agri_inputs.dict()
+    data = form_data.agri_inputs.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['ai'], data)
 
     # agri products
     data = form_data.agri_products
-    data = [agri_prods.dict() for agri_prods in data]
+    data = [agri_prods.model_dump() for agri_prods in data]
     [indiv_crop.update({"__id": fid}) for indiv_crop in data]
     DBQueries.insert_to_database(db, collection_names['ap'], data)
 
     # data['__id'] = fid
     # DBQueries.insert_to_database(db, collection_names['ap'], data)
 
     # livestock nums
-    data = form_data.livestock_nos.dict()
+    data = form_data.livestock_nos.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['ln'], data)
 
     # major probs
-    data = form_data.major_problems.dict()
+    data = form_data.major_problems.model_dump()
     data['__id'] = fid
     DBQueries.insert_to_database(db, collection_names['mp'], data)
 
@@ -240,5 +240,17 @@ def create_new_village(dbname,user_creds,response_result:FrontendResponseModel)-
     response_result['message'] = ['Authenticated','Village name added']
     response_result["data"]={}
 
+def get_resp_id_on_date(dbname,coll_name,date,response_result:FrontendResponseModel)->list:
+    """Wrapper function to get the list of users who have filled the form on a given date.
+    """
+    start_date=datetime.strptime(f"{date} 00:00:00","%d-%m-%Y %H:%M:%S")
+    end_date=datetime.strptime(f"{date} 23:59:59","%d-%m-%Y %H:%M:%S")
+
+    cursor=DBQueries.filtered_db_search(dbname,coll_name,['_id','__id','timestamp','volunteer_id'],
+                                        timestamp={"$gte":start_date,"$lte":end_date})
+    resp_data=[docs["resp_id"] for docs in cursor if docs["resp_id"]]
 
+    response_result['status'] = 'success'
+    response_result['message'] = ['Authenticated']
 
+    return resp_data

tests/intended_responses/respid_data.json

@@ -0,0 +1,23 @@
+{
+  "user_ids": [
+    "123412341234",
+    "012345678999",
+    "121212121212",
+    "121212121212",
+    "123412341235",
+    "123456789101",
+    "VIV1234567",
+    "102192837465",
+    "ZAS12345",
+    "111111115",
+    "111111116",
+    "111111117",
+    "111111118",
+    "111111119",
+    "873912790318",
+    "448003990988988",
+    "516329145259208",
+    "327247887001101",
+    "645574464113240"
+  ]
+}

tests/login_utils.py

@@ -33,4 +33,4 @@ def query_post(url, headers, data):
 
 
 with open("tests/intended_responses/postdata.json", 'r') as f:
-    data = json.load(f)
+    data = json.load(f)    

tests/test_get_resp_id.py

@@ -0,0 +1,81 @@
+import os
+import json
+import unittest
+import requests
+from login_utils import get_access_token, BASE_URL
+
+class MyGetRespIDTestCase(unittest.TestCase):
+    url=BASE_URL+"/api/get_respid_list"
+
+    with open("tests/intended_responses/respid_data.json","r") as f:
+        data=json.load(f)
+
+    def test_get_respid_owner_valid_village(self):
+        signincred = {
+            "AADHAR_NO": f"{os.environ['ADMIN_ID']}",
+            "password": f"{os.environ['ADMIN_PWD']}",
+            "village_name": f"{os.environ['ADMIN_VILLAGE_NAME']}",
+            "role": f"{os.environ['OWNER_ROLE']}"
+        }
+        params={"date":"27-02-2023","village_name":"Sehore"}
+        headers={
+            "accept":"application/json",
+            "Authorization": f"Bearer {get_access_token(signincred)}",
+            "Content-Type": "application/json",
+        }
+        response=requests.get(url=self.url,params=params,headers=headers)
+        self.assertEqual(response.json()["data"],self.data)
+
+    def test_get_respid_owner_invalid_village(self):
+        signincred = {
+            "AADHAR_NO": f"{os.environ['ADMIN_ID']}",
+            "password": f"{os.environ['ADMIN_PWD']}",
+            "village_name": f"{os.environ['ADMIN_VILLAGE_NAME']}",
+            "role": f"{os.environ['OWNER_ROLE']}"
+        }
+        params={"date":"27-02-2023","village_name":"villageDoesNotExist"}
+        headers={
+            "accept":"application/json",
+            "Authorization": f"Bearer {get_access_token(signincred)}",
+            "Content-Type": "application/json",
+        }
+        response=requests.get(url=self.url,params=params,headers=headers)
+        self.assertEqual(response.status_code,400)
+
+    def test_get_respid_admin(self):
+        signincred = {
+            "AADHAR_NO": f"{os.environ['ADMIN_ID']}",
+            "password": f"{os.environ['ADMIN_PWD']}",
+            "village_name": f"{os.environ['ADMIN_VILLAGE_NAME']}",
+            "role": f"{os.environ['ADMIN_ROLE']}"
+        }
+
+        params={"date":"27-02-2023","village_name":"None"}   
+        headers = {
+            "accept": "application/json",
+            "Authorization": f"Bearer {get_access_token(signincred)}",
+            "Content-Type": "application/json",
+        }
+        response=requests.get(url=self.url,params=params,headers=headers)
+        self.assertEqual(response.json()["data"],self.data)
+
+    def test_get_respid_user(self):
+        signincred = {
+            "AADHAR_NO": f"{os.environ['ADMIN_ID']}",
+            "password": f"{os.environ['ADMIN_PWD']}",
+            "village_name": f"{os.environ['ADMIN_VILLAGE_NAME']}",
+            "role": f"{os.environ['USER_ROLE']}"
+        }
+        params={"date":"02-03-2023","village_name":"None"}
+        headers={
+            "accept":"application/json",
+            "Authorization": f"Bearer {get_access_token(signincred)}",
+            "Content-Type": "application/json",
+        }
+        response=requests.get(url=self.url,params=params,headers=headers)
+        self.assertEqual(response.status_code,401)
+
+
+
+if __name__=="__main__":
+    unittest.main()