Skip to content

Commit 96aa859

Browse files
hjsbloggerhjsblogger
committed
Addition of FastAPI example code
1 parent e9bccc8 commit 96aa859

File tree

4 files changed

+156
-0
lines changed

4 files changed

+156
-0
lines changed

‎tests/fastAPI/app/main.py

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
import os
2+
import ssl
3+
import certifi
4+
import aiohttp
5+
from fastapi import FastAPI, HTTPException
6+
from dotenv import load_dotenv
7+
8+
load_dotenv()
9+
10+
user_name = os.getenv('LT_USERNAME')
11+
api_key = os.getenv('LT_ACCESS_KEY')
12+
13+
14+
app = FastAPI()
15+
16+
async def get_lambdatest_all_builds(session):
17+
url = f"https://{user_name}:{api_key}@api.lambdatest.com/automation/api/v1/builds?limit=50"
18+
headers = {"accept": "application/json"}
19+
async with session.get(url, headers=headers) as response:
20+
if response.status == 200:
21+
builds_data = await response.json()
22+
print(builds_data)
23+
return builds_data
24+
else:
25+
raise HTTPException(status_code=response.status, detail=await response.text())
26+
27+
async def get_lambdatest_all_sessions(session):
28+
url = f"https://{user_name}:{api_key}@api.lambdatest.com/automation/api/v1/sessions"
29+
headers = {"accept": "application/json"}
30+
async with session.get(url, headers=headers) as response:
31+
if response.status == 200:
32+
sessions_data = await response.json()
33+
print(sessions_data)
34+
return sessions_data
35+
else:
36+
raise HTTPException(status_code=response.status, detail=await response.text())
37+
38+
@app.get("/builds/")
39+
async def fetch_builds():
40+
async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl.create_default_context(cafile=certifi.where()))) as session:
41+
builds_data = await get_lambdatest_all_builds(session)
42+
dashboard_urls = [build['dashboard_url'] for build in builds_data.get('data', [])]
43+
return {"dashboard_urls": dashboard_urls}
44+
45+
@app.get("/sessions/")
46+
async def fetch_sessions():
47+
async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl.create_default_context(cafile=certifi.where()))) as session:
48+
sessions_data = await get_lambdatest_all_sessions(session)
49+
session_names = [session['build_name'] for session in sessions_data.get('data', [])]
50+
return {"session_names": session_names}
51+
52+
if __name__ == "__main__":
53+
import uvicorn
54+
uvicorn.run(app, host="0.0.0.0", port=8000)

‎tests/fastAPI/app/test_main.py

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
import pytest
2+
import aiohttp
3+
from fastapi.testclient import TestClient
4+
from tests.fastAPI.main import app
5+
6+
client = TestClient(app)
7+
8+
@pytest.mark.asyncio
9+
async def test_fetch_builds():
10+
async with aiohttp.ClientSession() as session:
11+
response = client.get("/builds/")
12+
assert response.status_code == 200
13+
data = response.json()
14+
assert "dashboard_urls" in data
15+
assert isinstance(data["dashboard_urls"], list)
16+
17+
@pytest.mark.asyncio
18+
async def test_fetch_sessions():
19+
async with aiohttp.ClientSession() as session:
20+
response = client.get("/sessions/")
21+
assert response.status_code == 200
22+
data = response.json()
23+
assert "session_names" in data
24+
assert isinstance(data["session_names"], list)

‎tests/fastAPI/main.py

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
import os
2+
import ssl
3+
import certifi
4+
import aiohttp
5+
from fastapi import FastAPI, HTTPException
6+
from dotenv import load_dotenv
7+
8+
load_dotenv()
9+
10+
user_name = os.getenv('LT_USERNAME')
11+
api_key = os.getenv('LT_ACCESS_KEY')
12+
13+
app = FastAPI()
14+
15+
async def get_lambdatest_all_builds(session):
16+
url = f"https://{user_name}:{api_key}@api.lambdatest.com/automation/api/v1/builds?limit=2000"
17+
headers = {"accept": "application/json"}
18+
async with session.get(url, headers=headers) as response:
19+
if response.status == 200:
20+
builds_data = await response.json()
21+
# print(builds_data)
22+
return builds_data
23+
else:
24+
raise HTTPException(status_code=response.status, detail=await response.text())
25+
26+
async def get_lambdatest_all_sessions(session):
27+
url = f"https://{user_name}:{api_key}@api.lambdatest.com/automation/api/v1/sessions"
28+
headers = {"accept": "application/json"}
29+
async with session.get(url, headers=headers) as response:
30+
if response.status == 200:
31+
sessions_data = await response.json()
32+
# print(builds_data)
33+
return sessions_data
34+
else:
35+
raise HTTPException(status_code=response.status, detail=await response.text())
36+
37+
@app.get("/builds/")
38+
async def fetch_builds():
39+
async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl.create_default_context(cafile=certifi.where()))) as session:
40+
builds_data = await get_lambdatest_all_builds(session)
41+
dashboard_urls = [build['dashboard_url'] for build in builds_data.get('data', [])]
42+
return {"dashboard_urls": dashboard_urls}
43+
44+
@app.get("/sessions/")
45+
async def fetch_sessions():
46+
async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl.create_default_context(cafile=certifi.where()))) as session:
47+
sessions_data = await get_lambdatest_all_sessions(session)
48+
session_names = [session['build_name'] for session in sessions_data.get('data', [])]
49+
return {"session_names": session_names}
50+
51+
if __name__ == "__main__":
52+
import uvicorn
53+
uvicorn.run(app, host="0.0.0.0", port=8000)

‎tests/fastAPI/test_main.py

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
import pytest
2+
import aiohttp
3+
from fastapi.testclient import TestClient
4+
from tests.fastAPI.main import app
5+
# from tests.FastAPI.main import app
6+
7+
client = TestClient(app)
8+
9+
@pytest.mark.asyncio
10+
async def test_fetch_builds():
11+
async with aiohttp.ClientSession() as session:
12+
response = client.get("/builds/")
13+
assert response.status_code == 200
14+
data = response.json()
15+
assert "dashboard_urls" in data
16+
assert isinstance(data["dashboard_urls"], list)
17+
18+
@pytest.mark.asyncio
19+
async def test_fetch_sessions():
20+
async with aiohttp.ClientSession() as session:
21+
response = client.get("/sessions/")
22+
assert response.status_code == 200
23+
data = response.json()
24+
assert "session_names" in data
25+
assert isinstance(data["session_names"], list)

0 commit comments

Comments
 (0)