|
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 client(): |
|
client = TestClient(app) |
|
yield client |
|
@pytest.fixture |
|
def inviter(): |
|
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) |
|
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 invitee(): |
|
url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword?key=AIzaSyArSoK9Wx9Hpe1R9ZywuLEIMVjCtHjO8Os" |
|
|
|
payload = json.dumps({ |
|
"email": "test2@gmail.com", |
|
"password": "testing2", |
|
"returnSecureToken": True |
|
}) |
|
headers = { |
|
'Content-Type': 'application/json' |
|
} |
|
response = requests.request("POST", url, headers=headers, data=payload) |
|
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): |
|
|
|
db = firestore.client(app=firebase_app) |
|
friend_request_ref = db.collection('friend_request') |
|
|
|
query = friend_request_ref.where(filter=FieldFilter("inviter", "==", inviter['id'])) |
|
docs = query.stream() |
|
for doc in docs: |
|
doc.reference.delete() |
|
|
|
user_ref = db.collection("user") |
|
user_ref.document(inviter['id']).delete() |
|
|
|
payload = '' |
|
headers = { |
|
'Content-Type': 'application/json', |
|
} |
|
response = client.request("POST", 'friend_request', headers=headers, data=payload) |
|
assert response.status_code == 403 |
|
|
|
payload = '' |
|
headers = { |
|
'Content-Type': 'application/json', |
|
'Authorization': 'Bearer amksckmasckmafvqnwfniqoniofv' |
|
} |
|
response = client.request("POST", 'friend_request', headers=headers, data=payload) |
|
assert response.status_code == 401 |
|
|
|
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 |
|
|
|
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) |
|
|
|
assert response.status_code == 200 |
|
result = mmcv.imfrombytes(response.read()) |
|
|
|
assert result.shape[2] == 3 |
|
|
|
mmcv.imwrite(result, "qrcode.jpg") |
|
|
|
|
|
user_ref.document(invitee['id']).delete() |
|
|
|
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 |
|
|
|
|
|
user_ref.document(invitee['id']).set({"deviceId": deviceId}) |
|
|
|
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 |
|
|
|
user_ref.document(inviter['id']).delete() |
|
user_ref.document(invitee['id']).delete() |
|
|