53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
import copy
|
|
import unittest
|
|
|
|
from scrapy.exceptions import DropItem
|
|
|
|
from FourmiCrawler import pipelines, spider, items
|
|
|
|
|
|
class TestPipelines(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.testItem = items.Result()
|
|
|
|
def test_NonePipeline(self):
|
|
#Testing the pipeline that replaces the None values in items.
|
|
self.testItem["value"] = "abc"
|
|
pipe = pipelines.RemoveNonePipeline()
|
|
processed = pipe.process_item(self.testItem, spider.FourmiSpider())
|
|
|
|
self.assertTrue(processed["value"] == "abc")
|
|
|
|
for key in self.testItem:
|
|
self.assertIsNotNone(processed[key])
|
|
if key is not "value":
|
|
self.assertIs(processed[key], "")
|
|
|
|
def test_DuplicatePipeline(self):
|
|
#Testing the pipeline that removes duplicates.
|
|
self.testItem["attribute"] = "test"
|
|
self.testItem["value"] = "test"
|
|
self.testItem["conditions"] = "test"
|
|
|
|
pipe = pipelines.DuplicatePipeline()
|
|
self.assertEqual(pipe.process_item(self.testItem, spider.FourmiSpider()), self.testItem)
|
|
self.assertRaises(DropItem, pipe.process_item, self.testItem, spider.FourmiSpider())
|
|
|
|
otherItem = copy.deepcopy(self.testItem)
|
|
otherItem["value"] = "test1"
|
|
self.assertEqual(pipe.process_item(otherItem, spider.FourmiSpider()), otherItem)
|
|
|
|
def test_AttributeSelection(self):
|
|
#Testing the pipeline that selects attributes.
|
|
item1 = copy.deepcopy(self.testItem)
|
|
item2 = copy.deepcopy(self.testItem)
|
|
|
|
item1["attribute"] = "abd"
|
|
item2["attribute"] = "abc"
|
|
|
|
s = spider.FourmiSpider(selected_attributes=["a.d"])
|
|
pipe = pipelines.AttributeSelectionPipeline()
|
|
|
|
self.assertEqual(pipe.process_item(item1, s), item1)
|
|
self.assertRaises(DropItem, pipe.process_item, item2, s) |