diff --git a/stix2/test/test_data_sources.py b/stix2/test/test_data_sources.py index 3327ca9..d798a0c 100644 --- a/stix2/test/test_data_sources.py +++ b/stix2/test/test_data_sources.py @@ -1,7 +1,7 @@ import pytest from taxii2client import Collection -from stix2 import Filter, MemorySource +from stix2 import Filter, MemorySink, MemorySource from stix2.sources import (CompositeDataSource, DataSink, DataSource, DataStore, make_id, taxii) from stix2.sources.filters import apply_common_filters @@ -20,11 +20,6 @@ def collection(): return Collection(COLLECTION_URL, MockTAXIIClient()) -@pytest.fixture -def ds(): - return DataSource() - - IND1 = { "created": "2017-01-27T13:49:53.935Z", "id": "indicator--d81f86b9-975b-bc0b-775e-810c5ad45a4f", @@ -127,21 +122,17 @@ STIX_OBJS1 = [IND1, IND2, IND3, IND4, IND5] def test_ds_abstract_class_smoke(): - ds1 = DataSource() - ds2 = DataSink() - ds3 = DataStore(source=ds1, sink=ds2) + with pytest.raises(TypeError): + DataStore() - with pytest.raises(NotImplementedError): - ds3.add(None) + with pytest.raises(TypeError): + DataStore.get() - with pytest.raises(NotImplementedError): - ds3.all_versions("malware--fdd60b30-b67c-11e3-b0b9-f01faf20d111") + with pytest.raises(TypeError): + DataSource() - with pytest.raises(NotImplementedError): - ds3.get("malware--fdd60b30-b67c-11e3-b0b9-f01faf20d111") - - with pytest.raises(NotImplementedError): - ds3.query([Filter("id", "=", "malware--fdd60b30-b67c-11e3-b0b9-f01faf20d111")]) + with pytest.raises(TypeError): + DataSink() def test_ds_taxii(collection): @@ -177,7 +168,8 @@ def test_parse_taxii_filters(): assert taxii_filters == expected_params -def test_add_get_remove_filter(ds): +def test_add_get_remove_filter(): + ds = taxii.TAXIICollectionSource(collection) # First 3 filters are valid, remaining properties are erroneous in some way valid_filters = [ @@ -226,7 +218,7 @@ def test_add_get_remove_filter(ds): ds.filters.update(valid_filters) -def test_apply_common_filters(ds): +def test_apply_common_filters(): stix_objs = [ { "created": "2017-01-27T13:49:53.997Z", @@ -374,35 +366,35 @@ def test_apply_common_filters(ds): assert len(resp) == 0 -def test_filters0(ds): +def test_filters0(): # "Return any object modified before 2017-01-28T13:49:53.935Z" resp = list(apply_common_filters(STIX_OBJS2, [Filter("modified", "<", "2017-01-28T13:49:53.935Z")])) assert resp[0]['id'] == STIX_OBJS2[1]['id'] assert len(resp) == 2 -def test_filters1(ds): +def test_filters1(): # "Return any object modified after 2017-01-28T13:49:53.935Z" resp = list(apply_common_filters(STIX_OBJS2, [Filter("modified", ">", "2017-01-28T13:49:53.935Z")])) assert resp[0]['id'] == STIX_OBJS2[0]['id'] assert len(resp) == 1 -def test_filters2(ds): +def test_filters2(): # "Return any object modified after or on 2017-01-28T13:49:53.935Z" resp = list(apply_common_filters(STIX_OBJS2, [Filter("modified", ">=", "2017-01-27T13:49:53.935Z")])) assert resp[0]['id'] == STIX_OBJS2[0]['id'] assert len(resp) == 3 -def test_filters3(ds): +def test_filters3(): # "Return any object modified before or on 2017-01-28T13:49:53.935Z" resp = list(apply_common_filters(STIX_OBJS2, [Filter("modified", "<=", "2017-01-27T13:49:53.935Z")])) assert resp[0]['id'] == STIX_OBJS2[1]['id'] assert len(resp) == 2 -def test_filters4(ds): +def test_filters4(): # Assert invalid Filter cannot be created with pytest.raises(ValueError) as excinfo: Filter("modified", "?", "2017-01-27T13:49:53.935Z") @@ -410,21 +402,21 @@ def test_filters4(ds): "for specified property: 'modified'") -def test_filters5(ds): +def test_filters5(): # "Return any object whose id is not indicator--d81f86b8-975b-bc0b-775e-810c5ad45a4f" resp = list(apply_common_filters(STIX_OBJS2, [Filter("id", "!=", "indicator--d81f86b8-975b-bc0b-775e-810c5ad45a4f")])) assert resp[0]['id'] == STIX_OBJS2[0]['id'] assert len(resp) == 1 -def test_filters6(ds): +def test_filters6(): # Test filtering on non-common property resp = list(apply_common_filters(STIX_OBJS2, [Filter("name", "=", "Malicious site hosting downloader")])) assert resp[0]['id'] == STIX_OBJS2[0]['id'] assert len(resp) == 3 -def test_filters7(ds): +def test_filters7(): # Test filtering on embedded property stix_objects = list(STIX_OBJS2) + [{ "type": "observed-data", @@ -463,7 +455,7 @@ def test_filters7(ds): assert len(resp) == 1 -def test_deduplicate(ds): +def test_deduplicate(): unique = deduplicate(STIX_OBJS1) # Only 3 objects are unique @@ -483,14 +475,14 @@ def test_deduplicate(ds): def test_add_remove_composite_datasource(): cds = CompositeDataSource() - ds1 = DataSource() - ds2 = DataSource() - ds3 = DataSink() + ds1 = MemorySource() + ds2 = MemorySource() + ds3 = MemorySink() with pytest.raises(TypeError) as excinfo: cds.add_data_sources([ds1, ds2, ds1, ds3]) assert str(excinfo.value) == ("DataSource (to be added) is not of type " - "stix2.DataSource. DataSource type is ''") + "stix2.DataSource. DataSource type is ''") cds.add_data_sources([ds1, ds2, ds1]) diff --git a/stix2/test/test_filesystem.py b/stix2/test/test_filesystem.py index 7aaa3f5..1610615 100644 --- a/stix2/test/test_filesystem.py +++ b/stix2/test/test_filesystem.py @@ -340,7 +340,7 @@ def test_filesystem_object_with_custom_property(fs_store): fs_store.add(camp, True) - camp_r = fs_store.get(camp.id, True) + camp_r = fs_store.get(camp.id) assert camp_r.id == camp.id assert camp_r.x_empire == camp.x_empire @@ -354,7 +354,7 @@ def test_filesystem_object_with_custom_property_in_bundle(fs_store): bundle = Bundle(camp, allow_custom=True) fs_store.add(bundle, True) - camp_r = fs_store.get(camp.id, True) + camp_r = fs_store.get(camp.id) assert camp_r.id == camp.id assert camp_r.x_empire == camp.x_empire @@ -369,7 +369,7 @@ def test_filesystem_custom_object(fs_store): newobj = NewObj(property1='something') fs_store.add(newobj, True) - newobj_r = fs_store.get(newobj.id, True) + newobj_r = fs_store.get(newobj.id) assert newobj_r.id == newobj.id assert newobj_r.property1 == 'something' diff --git a/stix2/test/test_memory.py b/stix2/test/test_memory.py index 0603bf7..fbdcaf7 100644 --- a/stix2/test/test_memory.py +++ b/stix2/test/test_memory.py @@ -235,7 +235,7 @@ def test_memory_store_object_with_custom_property(mem_store): mem_store.add(camp, True) - camp_r = mem_store.get(camp.id, True) + camp_r = mem_store.get(camp.id) assert camp_r.id == camp.id assert camp_r.x_empire == camp.x_empire @@ -249,7 +249,7 @@ def test_memory_store_object_with_custom_property_in_bundle(mem_store): bundle = Bundle(camp, allow_custom=True) mem_store.add(bundle, True) - bundle_r = mem_store.get(bundle.id, True) + bundle_r = mem_store.get(bundle.id) camp_r = bundle_r['objects'][0] assert camp_r.id == camp.id assert camp_r.x_empire == camp.x_empire @@ -265,6 +265,6 @@ def test_memory_store_custom_object(mem_store): newobj = NewObj(property1='something') mem_store.add(newobj, True) - newobj_r = mem_store.get(newobj.id, True) + newobj_r = mem_store.get(newobj.id) assert newobj_r.id == newobj.id assert newobj_r.property1 == 'something'