How to use in_message_path method in avocado

Best Python code snippet using avocado_python

test_harmony.py

Source:test_harmony.py

1import json2from shutil import rmtree3import sys4from tempfile import mkdtemp5from unittest import TestCase6from unittest.mock import patch7from os import environ8from pathlib import Path9from urllib.parse import urlsplit10from netCDF4 import Dataset11import pytest12import podaac.merger.harmony.cli13@pytest.mark.usefixtures("pass_options")14class TestMerge(TestCase):15    @classmethod16    def setUpClass(cls):17        cls.__test_path = Path(__file__).parent.resolve()18        cls.__data_path = cls.__test_path.joinpath('data')19        cls.__harmony_path = cls.__data_path.joinpath('harmony')20        cls.__output_path = Path(mkdtemp(prefix='tmp-', dir=cls.__data_path))21    @classmethod22    def tearDownClass(cls):23        if not cls.KEEP_TMP:  # pylint: disable=no-member24            rmtree(cls.__output_path)25    def test_service_invoke(self):26        in_message_path = self.__harmony_path.joinpath('message.json')27        in_message_data = in_message_path.read_text()28        in_message = json.loads(in_message_data)29        # test with both paged catalogs and un-paged catalogs30        for in_catalog_name in ['catalog.json', 'catalog0.json']:31            in_catalog_path = self.__harmony_path.joinpath('source', in_catalog_name)32            test_args = [33                podaac.merger.harmony.cli.__file__,34                '--harmony-action', 'invoke',35                '--harmony-input', in_message_data,36                '--harmony-source', str(in_catalog_path),37                '--harmony-metadata-dir', str(self.__output_path),38                '--harmony-data-location', self.__output_path.as_uri()39            ]40            test_env = {41                'ENV': 'dev',42                'OAUTH_CLIENT_ID': '',43                'OAUTH_UID': '',44                'OAUTH_PASSWORD': '',45                'OAUTH_REDIRECT_URI': '',46                'STAGING_PATH': '',47                'STAGING_BUCKET': ''48            }49            with patch.object(sys, 'argv', test_args), patch.dict(environ, test_env):50                podaac.merger.harmony.cli.main()51            out_catalog_path = self.__output_path.joinpath('catalog.json')52            out_catalog = json.loads(out_catalog_path.read_text())53            item_meta = next(item for item in out_catalog['links'] if item['rel'] == 'item')54            item_href = item_meta['href']55            item_path = self.__output_path.joinpath(item_href).resolve()56            # -- Item Verification --57            item = json.loads(item_path.read_text())58            properties = item['properties']59            # Accumulation method checks60            self.assertEqual(item['bbox'], [-4, -3, 4, 3])61            self.assertEqual(properties['start_datetime'], '2020-01-01T00:00:00+00:00')62            self.assertEqual(properties['end_datetime'], '2020-01-05T23:59:59+00:00')63            # -- Asset Verification --64            data = item['assets']['data']65            collection_name = in_message['sources'][0]['collection']66            # Sanity checks on metadata67            self.assertTrue(data['href'].endswith(f'/{collection_name}_merged.nc4'))68            self.assertEqual(data['title'], f'{collection_name}_merged.nc4')69            self.assertEqual(data['type'], 'application/x-netcdf4')70            self.assertEqual(data['roles'], ['data'])71            # -- subset_files Verification --72            file_list = [73                '2020_01_01_7f00ff_global.nc',74                '2020_01_02_3200ff_global.nc',75                '2020_01_03_0019ff_global.nc',76                '2020_01_04_0065ff_global.nc',77                '2020_01_05_00b2ff_global.nc'78            ]79            path = urlsplit(data['href']).path80            dataset = Dataset(path)81            subset_files = dataset['subset_files'][:].tolist()82            subset_files.sort()...

test_runner_sysinfo.py

Source:test_runner_sysinfo.py

...4from avocado.core.settings import settings5from avocado.plugins.runners.sysinfo import SysinfoRunner6class BasicTests(unittest.TestCase):7    """Basic unit tests for the RequirementPackageRunner class"""8    def in_message_path(self, messages, path, sysinfo_type="pre"):9        path = os.path.join("sysinfo", sysinfo_type, path)10        for message in messages:11            if message.get("path", "") == path:12                return True13        return False14    def test_pre(self):15        kwargs = {16            "sysinfo": {17                "commands": ["uptime", "dmidecode"],18                "files": ["/proc/version", "/proc/meminfo"],19            }20        }21        runnable = Runnable("sysinfo", "pre", **kwargs, config=settings.as_dict())22        runner = SysinfoRunner()23        status = runner.run(runnable)24        messages = []25        while True:26            try:27                messages.append(next(status))28            except StopIteration:29                break30        self.assertTrue(self.in_message_path(messages, "uptime"))31        self.assertTrue(self.in_message_path(messages, "dmidecode"))32        self.assertTrue(self.in_message_path(messages, "meminfo"))33        self.assertTrue(self.in_message_path(messages, "version"))34    def test_post_fail(self):35        kwargs = {36            "sysinfo": {37                "fail_commands": ["uptime", "dmidecode"],38                "fail_files": ["/proc/version", "/proc/meminfo"],39            },40            "test_fail": True,41        }42        runnable = Runnable("sysinfo", "post", **kwargs, config=settings.as_dict())43        runner = SysinfoRunner()44        status = runner.run(runnable)45        messages = []46        while True:47            try:48                messages.append(next(status))49            except StopIteration:50                break51        self.assertTrue(self.in_message_path(messages, "uptime", "post"))52        self.assertTrue(self.in_message_path(messages, "dmidecode", "post"))53        self.assertTrue(self.in_message_path(messages, "meminfo", "post"))...

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.