How to use task_result method in autotest

Best Python code snippet using autotest_python

test_task_result.py

Source:test_task_result.py

...4from src.Testing.Result.task_result import TaskResult5from src.Testing.Task.Similarity.cosine_similarity_task import CosineSimilarityTask6class TestTaskResult:7    @staticmethod8    def _create_enabled_task_result():9        task = CosineSimilarityTask("SimilarityTask", Path())10        task_result = TaskResult(task, True)11        case_result = CaseResult("Berlin", "Paris", "Paris", True)12        task_result.add_case_result(case_result)13        return task_result14    def test_has_results(self):15        task_result = TestTaskResult._create_enabled_task_result()16        assert task_result.has_results()17        task_result.finalize()18        assert task_result.has_results()19    def test_has_execution_duration(self):20        task_result = TestTaskResult._create_enabled_task_result()21        assert task_result.execution_duration() == 022        task_result.finalize()23        assert task_result.execution_duration() > 024    def test_pass_rate(self):25        task_result = TestTaskResult._create_enabled_task_result()26        assert task_result.pass_rate() == 027        case_result = CaseResult("Vienna", "London", "Paris", False)28        task_result.add_case_result(case_result)29        assert task_result.pass_rate() == 030        task_result.finalize()31        assert task_result.pass_rate() == 5032    def test_representation_contains_all_case_results(self):33        task_result = TestTaskResult._create_enabled_task_result()34        case_result = CaseResult("Vienna", "London", "Paris", False)35        task_result.add_case_result(case_result)36        task_result.finalize()37        assert len(str(task_result).split("\n")) == 338        assert task_result.__str__() == task_result.__repr__()39    def test_disabled_representation_starts_with_prefix(self):40        task = CosineSimilarityTask("SimilarityTask", Path())41        task_result = TaskResult(task, False)42        assert not task_result.enabled43        assert str(task_result).startswith(TaskResult.DISABLED_PREFIX)44        assert len(str(task_result).split("\n")) == 145    def test_empty_result(self):46        task = CosineSimilarityTask("SimilarityTask", Path())47        task_result = TaskResult(task, True)48        task_result.finalize()49        assert not task_result.has_results()50        assert task_result.pass_rate() == 0.051        assert task_result.execution_duration() == 052        assert len(str(task_result)) > 053        assert len(str(task_result).split("\n")) == 154    def test_add_case_result_raises_if_finalized(self):55        task = CosineSimilarityTask("SimilarityTask", Path())56        task_result = TaskResult(task, True)57        task_result.finalize()58        with pytest.raises(Exception):59            task_result.add_case_result(CaseResult("input", "expected", "actual", False))60    def test_finalize_raises_if_finalized(self):61        task = CosineSimilarityTask("SimilarityTask", Path())62        task_result = TaskResult(task, True)63        task_result.finalize()64        with pytest.raises(Exception):65            task_result.finalize()66    def test_finalize_returns_self(self):67        task_result = self._create_enabled_task_result()...

app.py

Source:app.py

1from celery.result import AsyncResult2from fastapi import FastAPI, File, UploadFile, Request3from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse4from fastapi.staticfiles import StaticFiles5from fastapi.templating import Jinja2Templates6from fastapi.middleware.cors import CORSMiddleware7from io import BytesIO8import shitfaced9import database10from worker import create_task11app = FastAPI()12app.mount("/static", StaticFiles(directory="static"), name="static")13templates = Jinja2Templates(directory="templates")14app.add_middleware(15    CORSMiddleware,16    allow_origins=shitfaced.ALLOWED_HOSTS,17    allow_credentials=True,18    allow_methods=["*"],19    allow_headers=["*"],20)21app.add_event_handler("shutdown", database.shutdown_mongo)22@app.get('/', response_class=HTMLResponse)23async def root_get(request: Request):24    """Return the home page"""25    return templates.TemplateResponse("index.html", {'request': request})26@app.get("/tasks/{task_id}")27def get_status(task_id):28    """Return the status of a task"""29    shitfaced.debugLog(f"Task ID : {task_id}")30    task_result = AsyncResult(task_id)31    shitfaced.debugLog(f"Task Status : {task_result.status}")32    shitfaced.debugLog(f"Task Result : {task_result.result}")33    result = {34        "task_id": task_id,35        "task_status": task_result.status,36        "task_result": str(task_result.result)37    }38    return JSONResponse(result)39@app.get('/get_shitfaced/{mongo_id}')40async def get_shitfaced(mongo_id):41    """Return the actual image file"""42    def iterimage(image):43        with BytesIO(image) as file_like:44            yield from file_like45    bytes_data = await database.get_shitface_image(mongo_id, database.db)46    return StreamingResponse(iterimage(bytes_data))47@app.post('/upload', status_code=201)48async def upload_file(request: Request, file: UploadFile = File(...)):49    """Process the uploaded file"""50    if file and shitfaced.allowed_file(file.filename):51        # The image file seems valid! Detect faces and return the result.52        content = await file.read()53        record_id = await database.create_shitface(content, file.filename, file.content_type, request.headers, database.db)54        shitfaced.debugLog(f'Record is : {record_id}')55        task_result = create_task.delay(str(record_id), shitfaced.DEBUG)56        shitfaced.debugLog(f"Image Result is : {task_result}")57        if not task_result:58            return JSONResponse({'error': 'balls'})59        return JSONResponse({'task_id': str(task_result)})60    else:...

orders.py

Source:orders.py

1from app.worker.tasks import celery2from app.worker.tasks import divide, get_order3from fastapi import APIRouter4from app.dependencies import get_service5from app.schemas.orders import Order6from fastapi.responses import JSONResponse7from celery.result import AsyncResult8router = APIRouter()9@router.post("")10async def combine_orders(11        order: Order12) -> Order:13    result = get_order.delay(order.requests, order.n_max)14    task_result = AsyncResult(result.id)15    while task_result.status != "SUCCESS":16        task_result = AsyncResult(task_result.id)17        result = {18            "task_id": task_result.id,19            "task_status": task_result.status,20            "task_result": task_result.result21        }...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.