oaebu_workflows.onix_workflow.tests.test_onix_workflow

Module Contents

Classes

TestOnixWorkflow

Functionally test the workflow

class oaebu_workflows.onix_workflow.tests.test_onix_workflow.TestOnixWorkflow(*args, **kwargs)

Bases: observatory.platform.observatory_environment.ObservatoryTestCase

Functionally test the workflow

onix_data

test_make_release(mock_sel_table_suffixes)

Tests that the make_release function works as intended

test_cleanup()

Tests the cleanup function of the workflow

test_dag_load()

Test that the DAG loads

test_dag_structure()

Tests that the dag structure is created as expected on dag load

test_create_and_load_aggregate_works_table(mock_bq_query)

test_crossref_API_calls()

Test the functions that query the crossref event and metadata APIs

test_get_onix_records(mock_bq_query)

test_crossref_transform()

Test the function that transforms the crossref events data

test_insert_into_schema()

Tests the instert_into_schema function

test_utility_functions()

Test the standalone functions in the Onix workflow that aren’t specifially tested in other classes

setup_fake_lookup_tables(settings_dataset_id, fixtures_dataset_id, release_date, bucket_name)

Create a new onix and subject lookup and country tables with their own dataset and table ids. Populate them with some fake data.

Parameters:

• settings_dataset_id (str) – The dataset to store the country table

• fixtures_dataset_id (str) – The dataset to store the lookup tables

• release_date (pendulum.DateTime) – The release/snapshot date

• bucket_name (str) – The GCP bucket name to load the data to

setup_input_data(data_partners, settings_dataset_id, fixtures_dataset_id, crossref_master_dataset_id, partner_dataset, onix_dataset_id, release_date, bucket_name)

Uploads the data partner fixtures to their respective GCP bucket and bigquery tables. Creates the partners based on the originals - but changes the dataset ids for tests. Create a new onix and subject lookup and country tables with their own dataset and table ids.

Parameters:

• data_partners (List[str]) – The list of data partners

• settings_dataset_id (str) – The dataset to store the country table

• fixtures_dataset_id (str) – The dataset to store the lookup tables

• partner_dataset (str) – The bigquery dataset ID to load the data partner tables to

• crossref_master_dataset_id (str) – The bigquery dataset ID of the master crossref table

• onix_dataaset – The Bigquery dataset ID to load the onix partner table to

• release_date (pendulum.DateTime) – The release/snapshot date of sharded tables

• bucket_name (str) – The name of the bucket to upload the jsonl files to

• onix_dataset_id (str) –

Returns:

The resulting OaebuPartners

Return type:

List[oaebu_workflows.oaebu_partners.OaebuPartner]

run_workflow_e2e(dry_run=False)

End to end test of the ONIX workflow

Parameters:

dry_run (bool) –

test_workflow_e2e()

Test that ONIX Workflow works as expected

test_e2e_dry()

Test that ONIX Workflow works as expected