simon987/od-database
1
0
Fork 0
mirror of https://github.com/simon987/od-database.git synced 2025-04-20 02:46:45 +00:00
Code Issues Projects Releases Wiki Activity

Removed unmaintained tests

Browse Source
This commit is contained in:
Simon 2018-06-26 19:07:14 -04:00
parent 8ea57967e6
commit f859bd3f8d
2 changed files with 0 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
test/test_crawl_server.py
View File

@ -1,48 +0,0 @@
from flask_testing import LiveServerTestCase
import json
import requests
from crawl_server.server import app
class CrawlServerTest(LiveServerTestCase):
headers = {
"Content-Type": "application/json"
}
HOST = "http://localhost:9999"
def create_app(self):
self.app = app
app.config['LIVESERVER_PORT'] = 9999
return app
def test_put_task(self):
payload = json.dumps({
"website_id": 123,
"url": "a",
"priority": 2,
"callback_type": "c",
"callback_args": '{"d": 4}'
})
requests.post(self.HOST + "/task/put", data=payload, headers=self.headers)
r = requests.get(self.HOST + "/task")
self.assertEqual(200, r.status_code)
result = json.loads(r.text)[0]
self.assertEqual(result["website_id"], 123)
self.assertEqual(result["url"], "a")
self.assertEqual(result["priority"], 2)
self.assertEqual(result["callback_type"], "c")
self.assertEqual(result["callback_args"], '{"d": 4}')
payload = json.dumps({"website_id": 1, "url": "", "priority": 1, "callback_type": "", "callback_args": "{}"})
r = requests.post(self.HOST + "/task/put", data=payload)
self.assertEqual(400, r.status_code)
r2 = requests.post(self.HOST + "/task/put", headers=self.headers, data=payload)
self.assertEqual(200, r2.status_code)

85
test/test_search.py
View File

@ -1,85 +0,0 @@
from unittest import TestCase
import time
import json
import os
from search.search import ElasticSearchEngine
class SearchTest(TestCase):
def setUp(self):
self.search = ElasticSearchEngine("od-database-test")
self.search.reset()
time.sleep(0.5)
def test_ping(self):
self.assertTrue(self.search.ping(), "Search engine not running")
def test_import_and_search(self):
files = [
{"name": "PaNopTicon", "size": 1000000000000000000, "path": "c/d", "mtime": 1528765672},
{"name": "BLAckwAter.Park", "size": 123, "path": "", "mtime": None},
{"name": "10'000 days", "size": -1, "path": "c", "mtime": 12345},
{"name": "Dead Racer", "size": 1000, "path": "Speed Machine [FLAC]", "mtime": 12345}
]
in_str = ""
for file in files:
in_str += json.dumps(file) + "\n"
self.search.import_json(in_str, 123)
time.sleep(2)
self.assertEqual(4, self.search.es.count(self.search.index_name, "file")["count"])
# Search for 'pan' in PaNopTicon and expect 1 result, a scroll id, and an highlight
page = self.search.search("pan")
self.assertIsNotNone(page["_scroll_id"])
self.assertEqual(1, page["hits"]["total"])
self.assertIsNotNone(page["hits"]["hits"][0]["highlight"]["name"])
# Search for 'park' and expect BLAckwAter.Park
page = self.search.search("park")
self.assertEqual(1, page["hits"]["total"])
# Search for fla and expect Dead Racer
page = self.search.search("fla")
self.assertEqual(1, page["hits"]["total"])
# Search for 10'000 and expect 10'000 days
page = self.search.search("10'000")
self.assertEqual(1, page["hits"]["total"])
def test_scroll(self):
files = [
{"name": "PaNopTicon", "size": 1000000000000000000, "path": "c/d", "mtime": 1528765672},
{"name": "BLAckwAter.Park", "size": 123, "path": "", "mtime": None},
{"name": "10'000 days", "size": -1, "path": "c", "mtime": 12345},
{"name": "Dead Racer", "size": 1000, "path": "Speed Machine [FLAC]", "mtime": 12345}
]
in_str = ""
for file in files:
in_str += json.dumps(file) + "\n"
self.search.import_json(in_str, 123)
time.sleep(2)
page = self.search.search("")
scroll_id = page["_scroll_id"]
# next page
next_page = self.search.scroll(scroll_id)
next_scroll_id = next_page["_scroll_id"]
self.assertIsNotNone(next_scroll_id)
# again
next_page2 = self.search.scroll(next_scroll_id)
self.assertIsNotNone(next_page2["_scroll_id"])
def test_invalid_scroll(self):
invalid_scroll = "blahblah"
self.assertIsNone(self.search.scroll(invalid_scroll))