Merge branch 'main' of https://github.com/HunyyDev/SE113.O12_KCPM

app/__init__.py

@@ -34,7 +34,7 @@ if not firebase_admin._apps:
         )
     )
 else:
-    firebase_app = firebase_admin._apps
+    firebase_app = firebase_admin.get_app()
 db = firestore.client()
 
 # LOAD NEO4J DB

app/constants.py

@@ -167,4 +167,4 @@ classNames = [
     "toothbrush",
 ]
 
-test_uid = "VAbtgG1IrWSOqoJgxe7u2yEqpIg2"
+deviceId = 'dvg7OhkORV6mYFlUuNKNaE:APA91bHp-SGj4uLyyI9zpEFGrQDrklonQff0bJjtQWhGrJ-tyd_1u5NZSenxEkJzDZ-lyYsZCvmIrZnwNeu9CetsuJNGAU0lkNePNngR-76ytR8ZLvqTOMyf5RHWexsxaPa9Rop05WMi'

app/custom_mmcv/test_custom_mmcv.py

@@ -0,0 +1,51 @@
+from app.custom_mmcv.color import color_val
+from app.custom_mmcv.main import imshow_det_bboxes
+from app.constants import classNames
+import mmcv
+import cv2
+import numpy as np
+import pytest
+class TestCustomMMCV():
+    def test_color_var(self):
+        assert color_val("black") == (0, 0, 0)
+        with pytest.raises(KeyError):
+            color_val("purple")
+        with pytest.raises(AssertionError):
+            color_val(300)
+        assert color_val(255) == (255,255,255)
+        with pytest.raises(AssertionError):
+            color_val(300)
+        assert color_val(np.array([20,255,40])) == (20,255,40)
+        with pytest.raises(AssertionError):
+            color_val(np.array([20,350,40]))
+        with pytest.raises(AssertionError):
+            color_val(np.array([0,200,40,40]))
+        with pytest.raises(AssertionError):
+            color_val(np.array([-30,0,40,40]))
+        with pytest.raises(AssertionError):
+            color_val(np.zeros((1,3)))
+        with pytest.raises(TypeError):
+            color_val(30.5)
+    def test_imshow_det_bboxes(self):
+        image = mmcv.imread('demo.jpg')
+        bboxes = np.ones((1,5))
+        labels = np.zeros(1, dtype=np.int32)
+        result = imshow_det_bboxes(image, bboxes, labels, class_names=classNames, bbox_color="red", text_color='red', show=False)
+        assert (result[1,1,:] == (0,0,255)).all()
+        with pytest.raises(AssertionError):
+            bboxes = np.ones((1,3))
+            labels = np.zeros(1, dtype=np.int32)
+            result = imshow_det_bboxes(image, bboxes, labels, class_names=classNames, bbox_color="red",text_color="red", show=False)
+        with pytest.raises(AssertionError):
+            bboxes = np.ones((1,7))
+            labels = np.zeros(1, dtype=np.int32)
+            result = imshow_det_bboxes(image, bboxes, labels, class_names=classNames, bbox_color="red",text_color="red", show=False)
+        with pytest.raises(AssertionError):
+            bboxes = np.ones((1,5))
+            labels = np.zeros(4, dtype=np.int32)
+            result = imshow_det_bboxes(image, bboxes, labels, class_names=classNames, bbox_color="red",text_color="red", show=False)
+        with pytest.raises(AssertionError):
+            bboxes = np.ones((2,5))
+            labels = np.zeros(1, dtype=np.int32)
+            result = imshow_det_bboxes(image, bboxes, labels, class_names=classNames, bbox_color="red",text_color="red", show=False)
+        

app/routers/image.py

@@ -58,22 +58,3 @@ def inferenceImage(img, threshold: float, isRaw: bool = False):
         score_thr=threshold,
     )
 
-
-@router.websocket("/")
-async def websocketEndpoint(websocket: WebSocket, threshold: float = 0.3):
-    await websocket.accept()
-    try:
-        while True:
-            data = await websocket.receive_bytes()
-            try:
-                img = imfrombytes(data, cv2.IMREAD_COLOR)
-                bboxes, labels = inferenceImage(img, threshold, True)
-            except Exception as e:
-                logger.error(e)
-                bboxes, labels = [], []
-
-            await websocket.send_json(
-                {"bboxes": bboxes.tolist(), "labels": labels.tolist()}
-            )
-    except WebSocketDisconnect:
-        pass

app/routers/test_friend_request.py

@@ -1,8 +1,45 @@
+import os
 import pytest
 import json 
+import cv2
+import mmcv
+import firebase_admin
 import requests
+from fastapi.testclient import TestClient
+from firebase_admin import credentials
+from app.main import app
+from app.constants import deviceId
+from fastapi.routing import APIRoute
+from firebase_admin import firestore
+from google.cloud.firestore_v1.base_query import FieldFilter
+def endpoints():
+    endpoints = []
+    for route in app.routes:
+        if isinstance(route, APIRoute):
+            endpoints.append(route.path)
+    return endpoints
+def read_qr_code(filename):
+    """Read an image and read the QR code.
+    
+    Args:
+        filename (string): Path to file
+    
+    Returns:
+        qr (string): Value from QR code
+    """
+    try:
+        img = cv2.imread(filename)
+        detect = cv2.QRCodeDetector()
+        value, points, straight_qrcode = detect.detectAndDecode(img)
+        return value
+    except:
+        return
 @pytest.fixture
-def token_inviter():
+def client():
+    client = TestClient(app)
+    yield client
+@pytest.fixture
+def inviter():
     url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword?key=AIzaSyArSoK9Wx9Hpe1R9ZywuLEIMVjCtHjO8Os"
 
     payload = json.dumps({
@@ -14,10 +51,22 @@ def token_inviter():
     'Content-Type': 'application/json'
     }
     response = requests.request("POST", url, headers=headers, data=payload)
-    token = response.json()["idToken"]
-    yield token
+    data = response.json()
+    inviter = {"id": data['localId'], "token": data["idToken"]}
+    yield inviter
+@pytest.fixture()
+def firebase_app():
+    if "test" in firebase_admin._apps:
+        firebase_app = firebase_admin.get_app('test')
+        yield firebase_app
+    else:
+        firebase_app = firebase_admin.initialize_app(        credential= credentials.Certificate(
+            json.loads(os.environ.get("FIREBASE_CREDENTIALS"))
+        ),name='test')
+        yield firebase_app
+    
 @pytest.fixture()
-def token_invitee():
+def invitee():
     url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword?key=AIzaSyArSoK9Wx9Hpe1R9ZywuLEIMVjCtHjO8Os"
 
     payload = json.dumps({
@@ -29,5 +78,86 @@ def token_invitee():
     'Content-Type': 'application/json'
     }
     response = requests.request("POST", url, headers=headers, data=payload)
-    token = response.json()["idToken"]
-    yield token
+    data = response.json()
+    invitee = {"id": data['localId'], "token": data["idToken"]}
+    yield invitee
+class TestFriendRequest():
+    @pytest.mark.skipif("/friend_request" not in endpoints(),reason="Route not defined")
+    def test_post_friend(self, firebase_app, client, inviter, invitee):
+        # Call the firebase database
+        db = firestore.client(app=firebase_app)
+        friend_request_ref = db.collection('friend_request')
+        # Remove all the friend_request use for testing in the past
+        query = friend_request_ref.where(filter=FieldFilter("inviter", "==", inviter['id']))
+        docs = query.stream()
+        for doc in docs:
+            doc.reference.delete()
+        # Delete the user for safety-check 
+        user_ref = db.collection("user")
+        user_ref.document(inviter['id']).delete()
+        # Send request with no token
+        payload = ''
+        headers = {
+        'Content-Type': 'application/json',
+        }
+        response = client.request("POST", 'friend_request', headers=headers, data=payload)
+        assert response.status_code == 403
+        # Send request with false token
+        payload = ''
+        headers = {
+        'Content-Type': 'application/json',
+        'Authorization': 'Bearer amksckmasckmafvqnwfniqoniofv' 
+        }
+        response = client.request("POST", 'friend_request', headers=headers, data=payload)
+        assert response.status_code == 401
+        # Send request with unknown user
+        payload = ''
+        headers = {
+        'Content-Type': 'application/json',
+        'Authorization': 'Bearer ' + inviter['token']
+        }
+        response = client.request("POST", 'friend_request', headers=headers, data=payload)
+        assert response.status_code == 500
+        # Create request and re-send 
+        user_ref.document(inviter['id']).set({"deviceId": deviceId})
+        payload = ''
+        headers = {
+        'Content-Type': 'application/json',
+        'Authorization': 'Bearer ' + inviter['token']
+        }
+        response = client.request("POST", 'friend_request', headers=headers, data=payload)
+        # Check response status code
+        assert response.status_code == 200
+        result = mmcv.imfrombytes(response.read())
+        # Check returned QR image
+        assert result.shape[2] == 3
+        # Write image for later read
+        mmcv.imwrite(result, "qrcode.jpg")
+        # Now test for the invitee aka the one that scan QR code
+        # Delete invitee user (if existed)
+        user_ref.document(invitee['id']).delete()
+        # Test when the invitee is unknow user (no user entity in database)
+        request_id = read_qr_code("qrcode.jpg")
+        payload = ''
+        headers = {
+        'Content-Type': 'application/json',
+        'Authorization': 'Bearer ' + invitee['token']
+        }
+        response = client.request("PATCH", 'friend_request/' + request_id, headers=headers, data=payload)
+        assert response.status_code == 500
+
+        # Create invitee user
+        user_ref.document(invitee['id']).set({"deviceId": deviceId})
+        # Send request
+        request_id = read_qr_code("qrcode.jpg")
+        payload = ''
+        headers = {
+        'Content-Type': 'application/json',
+        'Authorization': 'Bearer ' + invitee['token']
+        }
+        response = client.request("PATCH", 'friend_request/' + request_id, headers=headers, data=payload)
+        assert response.status_code == 200
+        # Delete entity for next time test
+        user_ref.document(inviter['id']).delete()
+        user_ref.document(invitee['id']).delete()
+        

app/test_main.py

@@ -7,85 +7,9 @@ import os
 import site
 import shutil
 from fastapi.routing import APIRoute
+import firebase_admin
 from app import firebase_app
+from app.constants import deviceId
 import requests
-
-
-def get_site_packages():
-    # Get the list of directories
-    site_packages_dirs = site.getsitepackages()
-
-    # Find the "site-packages" directory in the list
-    for dir in site_packages_dirs:
-        if dir.endswith("site-packages"):
-            target_dir = dir
-            break
-        else:
-            target_dir = None
-    return target_dir
-
-
-def endpoints():
-    endpoints = []
-    for route in app.routes:
-        if isinstance(route, APIRoute):
-            endpoints.append(route.path)
-    return endpoints
-
-
-@pytest.fixture
-def client():
-    client = TestClient(app, "http://0.0.0.0:3000")
-    yield client
-
-
-@pytest.fixture
-def token():
-    url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword?key=AIzaSyArSoK9Wx9Hpe1R9ZywuLEIMVjCtHjO8Os"
-
-    payload = json.dumps(
-        {"email": "test@gmail.com", "password": "testing", "returnSecureToken": True}
-    )
-    headers = {"Content-Type": "application/json"}
-    response = requests.request("POST", url, headers=headers, data=payload)
-    token = response.json()["idToken"]
-    yield token
-
-
-class TestFireBaseAPI:
-    def test_get_me(self, client, token):
-        if "/me" not in endpoints():
-            pytest.skip("This route isn't defined")
-        else:
-            if token != "":
-                payload = ""
-                headers = {
-                    "accept": "application/json",
-                    "Authorization": "Bearer " + token,
-                }
-                response = client.request("get", "me", headers=headers, data=payload)
-                assert response.status_code == 200
-            payload = ""
-            headers = {
-                "accept": "application/json",
-                "Content-Type": "application/json",
-            }
-            response = client.request("get", "me", headers=headers, data=payload)
-            assert response.status_code == 403
-
-    def test_invitation(self, client, token):
-        if "/invitation" not in endpoints():
-            pytest.skip("This route isn't defined")
-        else:
-            payload = ""
-            headers = {
-                "accept": "application/json",
-                "Content-Type": "application/json",
-                "Authorization": "Bearer " + token,
-            }
-            response = client.request(
-                "post", "invitation", headers=headers, data=payload
-            )
-            assert response.status_code == 200
-            result = mmcv.imfrombytes(response.read())
-            assert result.shape[0] == 3
+from firebase_admin import firestore
+from firebase_admin import credentials