ScrapeYard/module_test.py at main · habitualdev/ScrapeYard · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
#!/usr/bin/python3
import os
import pytest
import threading
import importlib
import time
from manager.config import get_config
import redis
import json
from yaml import CLoader as Loader, CDumper as Dumper
import yaml


parsed_yaml = get_config()
redis_host = parsed_yaml['Databases']['Redis']['Host']
redis_port = parsed_yaml['Databases']['Redis']['Port']


def module_test(test_module):
    with open("ingest_test_report.txt", "a") as f:
        r = redis.Redis(host=redis_host, port=redis_port)
        try:
            r.info()
            f.write("Redis connection successful\n")
            f.write("\n")
            print("Redis connection successful\n")
        except:
            f.write("Redis database not accessible. Check Redis settings inside config.yaml, or if the database is running.\n")
            print("Redis database not accessible. Check Redis settings inside config.yaml, or if the database is running.\n")
            return False
        module = importlib.import_module("ingest." + test_module)
        try:
            type(module.QueryClass)
        except:
            f.write("No QueryClass class to instantiate. \n")
            print("No QueryClass class to instantiate. \n")
            return False
        thread = threading.Thread(target=module.QueryClass, daemon=True)
        time.sleep(1)
        thread.start()
        json_data = r.lpop("data")
        if json_data is None:
            f.write("No data returned. Not a critical Error, may just be a scraping error.")
            print("No data returned. Not a critical Error, may just be a scraping error.")
            return True
        else:
            f.write("Data returned/ingested:")
            f.write(" -- " + str(json_data)[1:])
            f.write("\n")
            f.write("JSON format test:")
            print("Data returned/ingested:")
            print(" -- ", str(json_data)[1:], "\n")
            print("JSON format test:")
            try:
                json.loads(json_data)
                f.write(" -- JSON parses properly \n")
                print(" -- JSON parses properly", "\n")
            except:
                f.write(" -- JSON not parsable. \n")
                print(" -- JSON not parsable.", "\n")
                return False
        f.write("Flushing Redis database of any testing data. \n")
        print("Flushing Redis database of any testing data.", "\n")
        r.flushall()
        return True


def test_ingest_module():
    try:
        os.remove("ingest_test_report.txt")
    except:
        print("No previous report, continuing...")
    with open("config.yaml", "r") as f:
        yaml_raw = f.read()
    yaml_parsed = yaml.load(yaml_raw, Loader=Loader)
    for mod in yaml_parsed["Modules"]["Ingest"]:
        with open("ingest_test_report.txt", "a") as f:
            f.write("Testing %s module...\n" % mod)
            print("Testing %s module...\n" % mod)
            assert module_test(mod)
            f.write("---------------------------------------------------------")
            print("---------------------------------------------------------")
        time.sleep(2)