I have the file structure:
PROJECT
- contract-metadata
-- init.py
-- test_function.py
this is my init.py:
import logging
import json
import azure.functions as func
#from FastAPIApp import app # Main API application
from azure.cosmos import CosmosClient
import fastapi
import os
app = fastapi.FastAPI()
client = CosmosClient.from_connection_string(os.environ["CONNECTIONSTRINGSETTING"])
database_name = "dmdb"
container_name = "DataContract"
database = client.get_database_client(database_name)
container = database.get_container_client(container_name)
logging.info(f"container: {container}")
@app.get("/marketplace/{V}")
async def get_data_by_version(V: str):
logging.info("all data contracts are coming....")
query = f''' SELECT c.name,c.version, c.Theme, c.title, c.description, udf.tsToDate(c._ts) as updated_at
FROM c
WHERE c.version = '{V}'
ORDER BY c._ts DESC
OFFSET 0 LIMIT 20
'''
items = list(container.query_items(query, enable_cross_partition_query=True))
logging.info(f"items: {items}")
response_body = json.dumps(items)
return func.HttpResponse(body=response_body, mimetype="application/json")
@app.get("/marketplace/{V}/themes")
async def get_existing_themes_by_version(V: str):
query = f"SELECT c.Theme FROM c WHERE c.version = '{V}'"
items = list(container.query_items(query, enable_cross_partition_query=True))
unique_themes = list(set(item["Theme"] for item in items))
response_body = json.dumps(unique_themes)
return func.HttpResponse(body=response_body, mimetype="application/json")
@app.get("/marketplace/{V}/{id}")
async def get_data_by_version_and_id(V: str, id: str):
query = f"SELECT * FROM c WHERE c.version = '{V}' AND c.id = '{id}'"
items = list(container.query_items(query, enable_cross_partition_query=True))
if not items:
return func.HttpResponse(
f"No item with version '{V}' and id '{id}' was found",
status_code=404
)
response_body = json.dumps(items[0])
return func.HttpResponse(body=response_body, mimetype="application/json")
async def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
return await func.AsgiMiddleware(app).handle_async(req, context)
this is my test_func.py:
from fastapi.testclient import TestClient
import azure.functions as func
import unittest
import pytest
from __init__ import (
app, main,
get_data_by_version,
get_existing_themes_by_version,
get_data_by_version_and_id
)
# Unit tests
client = TestClient(app)
def test_get_data_by_version():
expected_data = [
{"name": "contract1", "version": "V1", "Theme": "Theme1", "title": "Title1", "description": "Description1"}
]
response = client.get("/marketplace/V1")
assert response.status_code == 200
assert response.json() == expected_data
def test_get_existing_themes_by_version():
expected_themes = ["Theme1", "Theme2"]
response = client.get("/marketplace/V1/themes")
assert response.status_code == 200
assert response.json() == expected_themes
def test_get_data_by_version_and_id():
expected_data = {"name": "contract1", "version": "V1", "id": "123"}
response = client.get("/marketplace/V1/123")
assert response.status_code == 200
assert response.json() == expected_data
I could not import from init import ( app, main, get_data_by_version, get_existing_themes_by_version, get_data_by_version_and_id )
These functions and I could not import the pytest, neither. I installed but it did not work.
Could you regulate my test file according to my code?
I could not import from **init** import ( app, main, get_data_by_version, get_existing_themes_by_version, get_data_by_version_and_id ) These functions and I could not import the pytest, neither. I installed but it did not work.
In order to import the functions from init.py to test_func.py add this line of code in your test_func.py :-
import sys
from fastapi.testclient import TestClient
import azure.functions as func
import unittest
import pytest
import os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), 'CosmosTrigger1')))
import __init__
__init__.get_data_by_version
__init__.get_existing_themes_by_version
__init__.app
__init__.main
__init__.get_data_by_version_and_id
Here CosmosTrigger1 is the name of the folder which contains init.py and test_func.py and the functions from init.py is imported successfully, Refer below:-