Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/npm/tests/test_lister.py
| Show First 20 Lines • Show All 44 Lines • ▼ Show 20 Lines | class NpmIncrementalListerTester(HttpListerTesterBase, unittest.TestCase): | ||||
| @requests_mock.Mocker() | @requests_mock.Mocker() | ||||
| def test_is_within_bounds(self, http_mocker): | def test_is_within_bounds(self, http_mocker): | ||||
| # disable this test from HttpListerTesterBase as | # disable this test from HttpListerTesterBase as | ||||
| # it can not succeed for the npm lister due to the | # it can not succeed for the npm lister due to the | ||||
| # overriding of the string_pattern_check method | # overriding of the string_pattern_check method | ||||
| pass | pass | ||||
| def test_lister_npm_basic_listing(swh_listers, requests_mock_datadir): | def check_tasks(tasks): | ||||
| lister = swh_listers['npm'] | """Ensure tasks as of expected formats | ||||
| lister.run() | |||||
| r = lister.scheduler.search_tasks(task_type='load-npm') | |||||
| assert len(r) == 100 | |||||
| for row in r: | """ | ||||
| for row in tasks: | |||||
| logger.debug('row: %s', row) | logger.debug('row: %s', row) | ||||
| assert row['type'] == 'load-npm' | assert row['type'] == 'load-npm' | ||||
| # arguments check | # arguments check | ||||
| args = row['arguments']['args'] | args = row['arguments']['args'] | ||||
| assert len(args) == 2 | assert len(args) == 2 | ||||
| package = args[0] | package = args[0] | ||||
| url = args[1] | url = args[1] | ||||
| assert url == 'https://www.npmjs.com/package/%s' % package | assert url == 'https://www.npmjs.com/package/%s' % package | ||||
| # kwargs | # kwargs | ||||
| kwargs = row['arguments']['kwargs'] | kwargs = row['arguments']['kwargs'] | ||||
| meta_url = kwargs['package_metadata_url'] | meta_url = kwargs['package_metadata_url'] | ||||
| assert meta_url == 'https://replicate.npmjs.com/%s' % package | assert meta_url == 'https://replicate.npmjs.com/%s' % package | ||||
| assert row['policy'] == 'recurring' | assert row['policy'] == 'recurring' | ||||
| assert row['priority'] is None | assert row['priority'] is None | ||||
| def test_lister_npm_basic_listing(swh_listers, requests_mock_datadir): | |||||
| lister = swh_listers['npm'] | |||||
| lister.run() | |||||
| tasks = lister.scheduler.search_tasks(task_type='load-npm') | |||||
| assert len(tasks) == 100 | |||||
| check_tasks(tasks) | |||||
| def test_lister_npm_listing_pagination(swh_listers, requests_mock_datadir): | |||||
| lister = swh_listers['npm'] | |||||
| # Patch per page pagination | |||||
| lister.per_page = 10 + 1 | |||||
| lister.PATH_TEMPLATE = lister.PATH_TEMPLATE.replace( | |||||
| '&limit=1001', '&limit=%s' % lister.per_page) | |||||
| lister.run() | |||||
| tasks = lister.scheduler.search_tasks(task_type='load-npm') | |||||
| assert len(tasks) == 2 * 10 # only 2 files with 10 results each | |||||
| check_tasks(tasks) | |||||