KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use process_records method in localstack

Best Python code snippet using localstack_python

test_cli.py

Source:test_cli.py

...7import records8def test_process_records_file():9    """Processes a single file sorting by lastname ascending. Outputs as comma separated values."""10    str_io = io.StringIO()11    records.process_records([str(Path(__file__).parent / 'data' / 'test.csv')], ['0,ASC'], 'csv', str_io)12    with open(str(Path(__file__).parent / 'expected' / 'test_process_records_file'), 'rb') as expected_stream:13        assert str_io.getvalue().encode("ascii") == expected_stream.read()14def test_process_records_files():15    """Processes multiple files sorting by lastname ascending. Outputs as comma separated values."""16    files = [17        str(Path(__file__).parent / 'data' / 'test.csv'),18        str(Path(__file__).parent / 'data' / 'test.psv')19    ]20    str_io = io.StringIO()21    records.process_records(files, ['0,ASC'], 'csv', str_io)22    with open(str(Path(__file__).parent / 'expected' / 'test_process_records_files'), 'rb') as expected_stream:23        assert str_io.getvalue().encode("ascii") == expected_stream.read()24def test_process_records_sort_lastname():25    """Processes a file and sorts by a last name descending. Outputs as pipe separated values."""26    str_io = io.StringIO()27    records.process_records([str(Path(__file__).parent / 'data' / 'test.ssv')], ['0,DESC'], 'psv', str_io)28    with open(str(Path(__file__).parent / 'expected' / 'test_process_records_sort_lastname'), 'rb') as expected_stream:29        assert str_io.getvalue().encode("ascii") == expected_stream.read()30def test_process_records_sort_lastname_date():31    """Processes a file and sorts by a last name descending and date ascending. Outputs as comma separated values."""32    str_io = io.StringIO()33    expected_filename = str(Path(__file__).parent / 'expected' / 'test_process_records_sort_lastname_date')34    records.process_records([str(Path(__file__).parent / 'data' / 'test.ssv')], ['0,DESC', '4,ASC'], 'csv', str_io)35    with open(expected_filename, 'rb') as expected_stream:36        assert str_io.getvalue().encode("ascii") == expected_stream.read()37# Test inputs: https://raw.githubusercontent.com/danielmiessler/SecLists/master/Fuzzing/38def test_process_records_fuzz():39    """Fuzz test CLI parser against a variety of potentially difficult to handle inputs."""40    str_io = io.StringIO()41    with open(Path(__file__).parent / 'data' / 'big-list-of-naughty-strings.txt', 'rb') as in_stream:42        blns = in_stream.readlines()43        blns = [s for s in blns if not s.startswith(b'#') and len(s) > 0]44    for s in blns:45        with tempfile.NamedTemporaryFile('wb', suffix=".ssv", delete=False) as tmp_file:46            tmp_file.truncate()47            tmp_file.seek(0)48            tmp_file.write(s)49        records.process_records([tmp_file.name], ['0,DESC', '4,ASC'], 'csv', str_io)50        unlink(tmp_file.name)51def test_process_records_nonexistent_file():52    """Test error handling on non-existent / inaccessible files."""53    with pytest.raises(FileNotFoundError):54        records.process_records(['non-existent.csv'], ['0,ASC'], 'csv')55def test_process_records_bad_sort():56    """Test error handling on bad sorts."""57    with pytest.raises(ValueError):58        records.process_records([str(Path(__file__).parent / 'data' / 'test.ssv')], ['0,ZESC', '4,ASC'], 'csv')59def test_process_records_bad_output_format():60    """Test error handling on bad output formats."""61    with pytest.raises(ValueError):...

routes.py

Source:routes.py

1import json2import logging3from typing import List4from fastapi import APIRouter5from fastapi import Depends6from fastapi import HTTPException7from fastapi import status8from sqlalchemy.orm import Session9from . import service10from .. import models11from ..database import get_db12logger = logging.getLogger(__name__)13router = APIRouter()14@router.get(15    "",16    response_model=models.ViewProcessAll,17    response_model_exclude_none=True,18    summary="The request asks the back-end for available processes and returns detailed process descriptions, including parameters and return values. Processes are described using the Functio specification for language-agnostic process descriptions",19)20def view_process_all(db_session: Session = Depends(get_db)):21    process_records = service.get_process_all(db_session=db_session)22    if not process_records:23        raise HTTPException(status_code=404, detail=f"No processes have been created")24    response_data = {25        "processes": process_records,26        "links": [27            {28                "rel": "alternate",29                "href": "https://jeodpp.jrc.ec.europa.eu/services/openeo/processes",30                "type": "text/html",31                "title": "HTML version of the processes",32            },33        ],34    }35    return response_data36@router.get(37    "/{process_name}",38    summary="The request will ask the back-end for further details about a process specified by the identifier process_name",39)40def view_process_detail(process_name: str):41    process = service.get_process_by_id(process_name)42    if not process:43        raise HTTPException(44            status_code=404,45            detail=f"No process with id {process_name} have been created",46        )...

test_list.py

Source:test_list.py

1from .list import process_records, write_csv2import io3def test_process_records():4    records = [5        dict(name="puma", type="kitty cat", age="100"),6        dict(name="sir-quackers", type="duck", age="101"),7        dict(name="john", type="moose", age="102"),8    ]9    # test use of fields and no filters10    filtered = process_records(records, fields=["name", "age"], filter_expressions=[])11    assert len(filtered) == 312    assert filtered[1] == {"name": "sir-quackers", "age": "101"}13    # test use of filters and getting all fields14    filtered = process_records(records, fields=None, filter_expressions=["type=duck"])15    assert filtered == [records[1]]16    filtered = process_records(records, fields=None, filter_expressions=["type!=duck"])17    assert filtered == [records[0], records[2]]18def test_process_nested_records():19    records = [dict(a=dict(b="1", c="2"), d="3"), dict(a=dict(b="4", c="5"), d="6")]20    # test use of fields and no filters21    filtered = process_records(records, fields=["a.b"], filter_expressions=[])22    assert len(filtered) == 223    assert filtered == [{"a": {"b": "1"}}, {"a": {"b": "4"}}]24    # test use of filters and getting all fields25    filtered = process_records(records, fields=None, filter_expressions=["a.b=4"])26    assert filtered == [records[1]]27def test_write_csv():28    # test flattening of nested structures when writing csv29    f = io.StringIO()30    write_csv([{"a": {"b": "c1", "d": "e2"}}, {"a": {"d": "e2"}, "b": "c2"}], f)...

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.