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

How to use get_schema_path method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

executor.py

Source:executor.py

...22    program_path = write_program(program, exec_id)23    wasm_path = compile_c(program_path, function_name)24    program_args = ["--wasm", f'server/{wasm_path}',25                    "--data", data_path,26                    "--schema", get_schema_path(data_path),27                    "--operation", operation,28                    "--function", function_name,29                    "--input", list_as_csv(input_column_names)]30    if output_column_name is not None:31        program_args.extend(["--output", output_column_name,32                            "--outputType", output_column_type])33    args_str = ' '.join(program_args)34    gradle_command = ' '.join(['./gradlew', 'run', f'--args=\"{args_str}\"'])35    print(gradle_command)36    subprocess.Popen(37        gradle_command, stdout=subprocess.PIPE, cwd=JAVA_ROOT, shell=True)38    return exec_id39def compile_c(src_file, exported_function):40    wasm_file = replace_extension(src_file, ".wasm")41    print(f'Compiling {wasm_file}...')42    process = subprocess.Popen([43        EMCC, '--no-entry', src_file,44        '-o', wasm_file,45        f'-sEXPORTED_FUNCTIONS=_{exported_function}'])46    process.wait()47    if not exists(wasm_file):48        raise RuntimeError('Failed to compile WASM')49    return wasm_file50def get_result_if_exists(id):51    if exists(f'{USER_DATA_PATH}/{id}/_SUCCESS'):52        for name in glob(f'{USER_DATA_PATH}/{id}/part*json'):53            data = []54            with open(name, 'r') as f:55                count = 056                for line in f:57                    data.append(json.loads(line))58                    count += 159                    if count > 200:60                        break61            return data62def list_data_files():63    process = subprocess.Popen(64        ["ls", USER_DATA_PATH], stdout=subprocess.PIPE, stderr=subprocess.PIPE)65    stdout, stderr = process.communicate()66    all_file_names = stdout.decode('UTF-8').split()67    return list(filter(lambda s: '.csv' in s, all_file_names))68def get_schema(file_name: str):69    print(f'Get schema for file {file_name}')70    schema_path = get_schema_path(f"{USER_DATA_PATH}/{file_name}")71    with open(schema_path, "r") as f:72        schema_str = f.readline()73        print(schema_str)74        return [{"name": entry.split()[0], "type": entry.split()[1]} for entry in schema_str.split(',')]75def write_program(program: str, exec_id: str) -> str:76    path = f"{USER_CODE_PATH}/{exec_id}.c"77    with open(path, "w") as f:78        f.write(program)79    return path80def get_schema_path(data_path: str):81    return replace_extension(data_path, "_schema.txt")82def replace_extension(file_path: str, new_extension: str):83    idx = file_path.rfind('.')84    return file_path[:idx] + new_extension85def list_as_csv(inp: List[str]):...

test_validation.py

Source:test_validation.py

...5from trickster.validation import request_schema, compile_json_schema, get_schema_path6@pytest.mark.unit7class TestJsonSchemaValidation:8    def test_schema_path_returns_absolute_path(self):9        path = get_schema_path('route.schema.json')10        assert path.name == 'route.schema.json'11        assert path.is_absolute12        assert path.exists()13    def test_compiled_json_schema_is_cached(self):14        validator1 = compile_json_schema(get_schema_path('route.schema.json'))15        validator2 = compile_json_schema(get_schema_path('route.schema.json'))16        assert validator1 is validator217    def test_compile_valid_json_schema(self, tmpdir):18        schema = tmpdir.join('test.schema.json')19        schema.write('''{20            "$schema": "http://json-schema.org/draft-07/schema#",21            "type": "object",22            "properties": {"property": {"type": "number"}},23            "required": ["property"]24        }''')25        validator = compile_json_schema(Path(schema))26        validator({'property': 2})27    def test_compile_invalid_json_schema(self, tmpdir):28        schema = tmpdir.join('test.schema.json')29        schema.write('''{...

test_path.py

Source:test_path.py

...5    current_path = os.path.dirname(os.path.abspath(__file__))6    schema_path = os.path.join(current_path, "Schema")7    os.environ['SCHEMA_PATH'] = schema_path8    expected_result = schema_path9    actual_result = get_schema_path()10    os.environ.pop('SCHEMA_PATH')11    assert actual_result == [expected_result]12def test_get_schema_path_exist_not_exist():13    paths = get_schema_path()14    match = [each_path for each_path in paths if "/test/Schema2" or "/test/Schema" in each_path]...

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.