Currently the upstream test is performed by calling updated_advisories with 1 batch size for all the importers. This is inefficient. Given that we want to move to non batch based processing anyway, we'd need to replace this test.
Instead we can have a test that loops over all the data sources and calls a test_upstream() function with the data source.
The test_upstream() would have a basic if-elif (switch like) structure which would run test for the respective upstream. The default case could be to fail the test entirely effectively rejecting the importer without an upstream test.
Something like this:
def test_upstream(data_source):
if data_source == "DebianDataSource":
# perform Debian upstream test here
return
..
..
..
raise NoUpstreamTestException
Currently the upstream test is performed by calling
updated_advisorieswith 1 batch size for all the importers. This is inefficient. Given that we want to move to non batch based processing anyway, we'd need to replace this test.Instead we can have a test that loops over all the data sources and calls a
test_upstream()function with the data source.The
test_upstream()would have a basicif-elif(switch like) structure which would run test for the respective upstream. The default case could be to fail the test entirely effectively rejecting the importer without an upstream test.Something like this: