========================
Ascend.io Test Framework
========================
This package helps developers who are writing custom python for Ascend.io automated pipelines by providing a local
testing framework. Local testing speeds the development of python pipeline code. The local framework exercises the
code as if the code was running directly in the platform while giving you access to patching and mocking frameworks.
Documentation, including examples, is located in our `Ascend Developer Hub <https://developer.ascend.io>`_.
Example
------------
Here is a basic python transformation test case example. The python code under test is located in a file
with the name ``my_python_transform.py`` and imported with the name ``my_python_transform``. Other variables,
imports, and code are omitted for brevity::
@AscendPySparkTransform(spark=spark_session,
module=my_python_transform,
schema=input_schema,
data=[(123, 'NORMAL', today, today + datetime.timedelta(days=1))],
credentials=test_creds,
discover_schema=True,
patches=[patch('requests.post', return_value=Mock(status_code=200,
text='{"internalReportIds":"REPORT_A"}')),
patch('requests.get', return_value=Mock(status_code=200,
text='{"status":"SUCCESS", "downloadLink": "https://test.my.download"}')),
patch('pandas.read_csv', return_value=build_mock_csv()),
], )
def test_normal_loading_process_single_record(input_dataframe, transform_result: DataFrame, mock_results: List[Mock]):
"""Check that a normal call does the work properly.
Assert values are correct.
Assert mock services are called."""
assert input_dataframe
assert transform_result
assert transform_result.count() == 3
dataset = transform_result.collect()
# check field mapping
assert dataset[0]['CUSTOMER_ID'] == '101'
assert dataset[1]['CUSTOMER_ID'] == '102'
assert dataset[2]['CUSTOMER_ID'] == '103'
assert dataset[0]['YOUR_NAME'] == "customerName.one"
assert dataset[0]['THE_OBJECTIVE'] == "customerBudget.one"
assert dataset[0]['AD_ID'] == "tempId.one"
assert dataset[0]['AD_NAME'] == "myName.one"
assert dataset[0]['GEO_LOC'] == "geo_location.one"
assert dataset[0]['ORDER_ID'] == "orderId.test"
assert dataset[0]['ORDER_NAME'] == "orderName.test"
assert dataset[0]['DT'] == "__time.one"
assert dataset[0]['AUDIO_IMPRESSIONS'] == 1
assert transform_result.columns.__contains__('RUN_ID')
assert transform_result.columns.__contains__('REPORT_START_DT')
assert transform_result.columns.__contains__('REPORT_END_DT')
assert transform_result.columns.__contains__('record_number')
# check mocks were properly called
mock_results[0].assert_called_once()
mock_results[1].assert_called_once_with(f"https://custom.io/v1/async-query/REPORT_A",
headers={'agency': '12', 'x-api-key': 'key', 'Content-Type': 'application/json'})
mock_results[2].assert_called_once_with("https://test.my.download", header=0, skip_blank_lines=True)
Decorators are available for all types of Ascend python implementation strategies. Testing scenarios are only limited
by your creativity and desire to produce high quality code.
Download your pipelines using the `Ascend CLI <https://pypi.org/project/ascend-io-cli/>`_ like this::
ascend download dataflow MY_DATASERVICE MY_DATA_FLOW
Write some tests. When your test cases are complete, pushing the code to the platform is simple with
the `CLI <https://pypi.org/project/ascend-io-cli/>`_. For example::
ascend apply dataflow MY_DATASERVICE MY_DATA_FLOW
---------------
Read the Docs
---------------
* `Ascend Developer Hub <https://developer.ascend.io>`_
* `Ascend.io <https://www.ascend.io>`_
* `Ascend CLI <https://pypi.org/project/ascend-io-cli/>`_
Raw data
{
"_id": null,
"home_page": "https://www.ascend.io",
"name": "ascend-io-test",
"maintainer": "Ascend.io Engineering",
"docs_url": null,
"requires_python": ">=3.9,<4.0",
"maintainer_email": "support@ascend.io",
"keywords": "ascend,pipeline,data,automation,platform",
"author": "Ascend.io Engineering",
"author_email": "support@ascend.io",
"download_url": "https://files.pythonhosted.org/packages/32/ea/e16cda3257c3c70616e54a19aba2b00a1c9543378266224e6c3c7214ffef/ascend_io_test-0.9.7.tar.gz",
"platform": null,
"description": "========================\nAscend.io Test Framework\n========================\n\nThis package helps developers who are writing custom python for Ascend.io automated pipelines by providing a local\ntesting framework. Local testing speeds the development of python pipeline code. The local framework exercises the\ncode as if the code was running directly in the platform while giving you access to patching and mocking frameworks.\n\nDocumentation, including examples, is located in our `Ascend Developer Hub <https://developer.ascend.io>`_.\n\nExample\n------------\nHere is a basic python transformation test case example. The python code under test is located in a file\nwith the name ``my_python_transform.py`` and imported with the name ``my_python_transform``. Other variables,\nimports, and code are omitted for brevity::\n\n @AscendPySparkTransform(spark=spark_session,\n module=my_python_transform,\n schema=input_schema,\n data=[(123, 'NORMAL', today, today + datetime.timedelta(days=1))],\n credentials=test_creds,\n discover_schema=True,\n patches=[patch('requests.post', return_value=Mock(status_code=200,\n text='{\"internalReportIds\":\"REPORT_A\"}')),\n patch('requests.get', return_value=Mock(status_code=200,\n text='{\"status\":\"SUCCESS\", \"downloadLink\": \"https://test.my.download\"}')),\n patch('pandas.read_csv', return_value=build_mock_csv()),\n ], )\n def test_normal_loading_process_single_record(input_dataframe, transform_result: DataFrame, mock_results: List[Mock]):\n \"\"\"Check that a normal call does the work properly.\n Assert values are correct.\n Assert mock services are called.\"\"\"\n assert input_dataframe\n assert transform_result\n assert transform_result.count() == 3\n dataset = transform_result.collect()\n # check field mapping\n assert dataset[0]['CUSTOMER_ID'] == '101'\n assert dataset[1]['CUSTOMER_ID'] == '102'\n assert dataset[2]['CUSTOMER_ID'] == '103'\n assert dataset[0]['YOUR_NAME'] == \"customerName.one\"\n assert dataset[0]['THE_OBJECTIVE'] == \"customerBudget.one\"\n assert dataset[0]['AD_ID'] == \"tempId.one\"\n assert dataset[0]['AD_NAME'] == \"myName.one\"\n assert dataset[0]['GEO_LOC'] == \"geo_location.one\"\n assert dataset[0]['ORDER_ID'] == \"orderId.test\"\n assert dataset[0]['ORDER_NAME'] == \"orderName.test\"\n assert dataset[0]['DT'] == \"__time.one\"\n assert dataset[0]['AUDIO_IMPRESSIONS'] == 1\n assert transform_result.columns.__contains__('RUN_ID')\n assert transform_result.columns.__contains__('REPORT_START_DT')\n assert transform_result.columns.__contains__('REPORT_END_DT')\n assert transform_result.columns.__contains__('record_number')\n # check mocks were properly called\n mock_results[0].assert_called_once()\n mock_results[1].assert_called_once_with(f\"https://custom.io/v1/async-query/REPORT_A\",\n headers={'agency': '12', 'x-api-key': 'key', 'Content-Type': 'application/json'})\n mock_results[2].assert_called_once_with(\"https://test.my.download\", header=0, skip_blank_lines=True)\n\n\nDecorators are available for all types of Ascend python implementation strategies. Testing scenarios are only limited\nby your creativity and desire to produce high quality code.\n\nDownload your pipelines using the `Ascend CLI <https://pypi.org/project/ascend-io-cli/>`_ like this::\n\n ascend download dataflow MY_DATASERVICE MY_DATA_FLOW\n\nWrite some tests. When your test cases are complete, pushing the code to the platform is simple with\nthe `CLI <https://pypi.org/project/ascend-io-cli/>`_. For example::\n\n ascend apply dataflow MY_DATASERVICE MY_DATA_FLOW\n\n\n\n---------------\nRead the Docs\n---------------\n* `Ascend Developer Hub <https://developer.ascend.io>`_\n* `Ascend.io <https://www.ascend.io>`_\n* `Ascend CLI <https://pypi.org/project/ascend-io-cli/>`_\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "The Ascend Python Test Framework",
"version": "0.9.7",
"project_urls": {
"Ascend Developer Hub": "https://developer.ascend.io",
"Ascend.io": "https://www.ascend.io",
"Documentation": "https://developer.ascend.io",
"Homepage": "https://www.ascend.io"
},
"split_keywords": [
"ascend",
"pipeline",
"data",
"automation",
"platform"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "13ac7bb9e467faef6b189fb4abedf9ae2f0a816799119160e8ce92745bbba4e2",
"md5": "41a90b8c940f49df8d25614f4a03187e",
"sha256": "fda7ebff8ee66d3d67a1eb815786d74c8d268c592c44d1a9ea40ffea15c70a18"
},
"downloads": -1,
"filename": "ascend_io_test-0.9.7-py3-none-any.whl",
"has_sig": false,
"md5_digest": "41a90b8c940f49df8d25614f4a03187e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9,<4.0",
"size": 11875,
"upload_time": "2023-09-07T12:39:25",
"upload_time_iso_8601": "2023-09-07T12:39:25.798941Z",
"url": "https://files.pythonhosted.org/packages/13/ac/7bb9e467faef6b189fb4abedf9ae2f0a816799119160e8ce92745bbba4e2/ascend_io_test-0.9.7-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "32eae16cda3257c3c70616e54a19aba2b00a1c9543378266224e6c3c7214ffef",
"md5": "0cd1e081ec01816cb2cdaa85f35a26d4",
"sha256": "19241300dae92e9d204bd9f7e30292adb5f67cf153515cb21b27b6ff693be73d"
},
"downloads": -1,
"filename": "ascend_io_test-0.9.7.tar.gz",
"has_sig": false,
"md5_digest": "0cd1e081ec01816cb2cdaa85f35a26d4",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9,<4.0",
"size": 9313,
"upload_time": "2023-09-07T12:39:27",
"upload_time_iso_8601": "2023-09-07T12:39:27.329345Z",
"url": "https://files.pythonhosted.org/packages/32/ea/e16cda3257c3c70616e54a19aba2b00a1c9543378266224e6c3c7214ffef/ascend_io_test-0.9.7.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-09-07 12:39:27",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "ascend-io-test"
}
JSON