Spaces:
Runtime error
Runtime error
File size: 2,944 Bytes
c6a9f21 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
from fastapi.testclient import TestClient
from fastapi.routing import APIRoute
from app.main import app
from app import db
import os
import pytest
import requests
import json
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
@pytest.fixture
def client():
client = TestClient(app, "http://0.0.0.0:3000")
yield client
@pytest.fixture
def user():
url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword?key=" + os.environ.get("FIREBASE_API_KEY")
payload = json.dumps({
"email": "[email protected]",
"password": "testing",
"returnSecureToken": True
})
headers = {
'Content-Type': 'application/json'
}
response = requests.request("POST", url, headers=headers, data=payload)
data = response.json()
user = {"id": data['localId'], "token": data["idToken"]}
yield user
class TestVideoAPI:
@pytest.mark.skipif("/video" not in endpoints(),reason="Route not defined")
def test_video_API(self, user, client):
# Test when sent file is not a video
payload = {}
files=[
('file',('demo.jpg',open('demo.jpg','rb'),'application/octet-stream'))
]
headers = {
'Content-Type': 'application/json',
}
response = client.request("POST", 'video', headers=headers, data=payload, files=files)
assert response.status_code == 400
assert response.text == "File must be video"
# Test when no token is pass to route
payload = {}
files=[
('file',('demo.mp4',open('demo.mp4','rb'),'application/octet-stream'))
]
headers = {
'Content-Type': 'application/json',
}
response = client.request("POST", 'video', headers=headers, data=payload, files=files)
assert response.status_code == 400
assert response.text == "User not found"
# Test when all requirements have been fulfilled
payload = {}
files=[
('file',('demo.mp4',open('demo.mp4','rb'),'application/octet-stream'))
]
headers = {
'Content-Type': 'application/json',
'Authorization': "Bearer " + user['token']
}
response = client.request("POST", 'video', headers=headers, data=payload, files=files)
assert response.status_code == 200
artifactName = response.text
docs = db.collection("artifacts").where(filter = FieldFilter("name", '==', artifactName))
index = 0
for doc in docs:
# For each document in docs. Verify name and status of the artifact
index += 1
data = doc.get().to_dict()
assert data['name'] == artifactName
assert data['status'] == 'pending'
assert index == 1
|