Newer
Older
"""Unit tests for testing the application functionality."""
# TODO: TEST BLINKER & GUNICORN IMPORTS
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
import copy
import logging
import ntpath
import os
from collections import OrderedDict
from threading import Thread
from unittest.mock import patch, MagicMock, mock_open
from zipfile import ZipFile
import rapidjson as json
import re
import shutil
import string
import sys
import time
import unittest
from multiprocessing import Process
from unittest import TestLoader
from datetime import datetime
from typing import Dict, List, Tuple, Type
from conllu import TokenList
from flask import Flask
from flask.testing import FlaskClient
from gensim.models import Word2Vec
from lxml import etree
from networkx import MultiDiGraph, Graph
from requests import HTTPError
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import session
from werkzeug.exceptions import NotFound
from werkzeug.wrappers import Response
import csm
import mcserver
from csm.app import create_csm_app
from mcserver.app import create_app, db, start_updater, full_init
from mcserver.app.api.exerciseAPI import map_exercise_data_to_database
from mcserver.app.api.vectorNetworkAPI import add_edges, get_concept_network
from mcserver.app.models import Corpus, UpdateInfo, ResourceType, Exercise, FileType, ExerciseType, ExerciseData, \
NodeMC, LinkMC, GraphData, Phenomenon, CustomCorpus, AnnisResponse, Solution, DownloadableFile, Language, \
VocabularyCorpus, SolutionElement, TextComplexityMeasure, FrequencyAnalysis, CitationLevel, FrequencyItem, \
LearningResult, TextComplexity, Dependency, PartOfSpeech, Category, Choice, Actor, XapiStatement
from mcserver.app.services import AnnotationService, CorpusService, FileService, CustomCorpusService, DatabaseService, \
XMLservice, TextService
from mcserver.config import TestingConfig, Config
from mocks import Mocks, MockResponse, MockW2V
class McTestCase(unittest.TestCase):
"""The test suite for the main application."""
def mocked_requests_get(*args, **kwargs):
if TestingConfig.SIMULATE_CORPUS_NOT_FOUND:
return MockResponse(json.dumps(AnnisResponse().__dict__))
elif TestingConfig.SIMULATE_HTTP_ERROR:
raise HTTPError
elif TestingConfig.SIMULATE_EMPTY_GRAPH:
graph_data_raw: dict = dict(Mocks.annis_response_dict["graph_data_raw"])
graph_data_raw["nodes"] = []
return MockResponse(json.dumps(graph_data_raw))
url: str = args[0]
if url == Config.CTS_API_BASE_URL:
if kwargs['params']['request'] == 'GetCapabilities':
return MockResponse(Mocks.cts_capabilities_xml)
return MockResponse(Mocks.cts_reff_xml)
elif url.endswith(Config.SERVER_URI_CSM_SUBGRAPH):
return MockResponse(json.dumps(Mocks.annis_response_dict))
else:
gd: GraphData = AnnotationService.map_graph_data(Mocks.annis_response_dict["graph_data_raw"])
return MockResponse(json.dumps(gd.serialize()))
def mocked_requests_post(*args, **kwargs):
url: str = args[0]
if url.endswith(Config.SERVER_URI_TEXT_COMPLEXITY):
return MockResponse(Mocks.text_complexity_json_string)
else:
ed1: ExerciseData = AnnotationService.map_graph_data_to_exercise(
Mocks.annis_response_dict["graph_data_raw"],
"", [Solution(target=SolutionElement(
salt_id='salt:/urn:custom:latinLit:proiel.pal-agr.lat:1.1.1/doc1#sent159692tok1'))])
ed2: ExerciseData = AnnotationService.map_graph_data_to_exercise(
Mocks.annis_response_dict["graph_data_raw"],
"", [Solution(target=SolutionElement(
salt_id='salt:/urn:custom:latinLit:proiel.pal-agr.lat:1.1.1/doc1#sent159695tok10'))])
ed2.graph.nodes = ed2.graph.nodes[42:]
return MockResponse(json.dumps([ed1.serialize(), ed2.serialize()]))
def setUp(self):
"""Initializes the testing environment."""
self.start_time = time.time()
if os.path.exists(Config.GRAPH_DATABASE_DIR):
shutil.rmtree(Config.GRAPH_DATABASE_DIR)
patcher = patch.object(TextService, "init_stop_words_latin")
self.addCleanup(patcher.stop)
patcher.start()
self.app: Flask = create_app(TestingConfig)
self.app.logger.setLevel(logging.CRITICAL)
self.assertIsInstance(self.app, Flask)
self.app_context = self.app.app_context()
self.app_context.push()
self.app.testing = True
self.client: FlaskClient = self.app.test_client()
self.assertIsInstance(self.client, FlaskClient)
UpdateInfo.query.delete()
Corpus.query.delete()
def tearDown(self):
"""Finishes testing by removing the traces."""
db.session.remove()
db.drop_all()
self.app_context.pop()
print("{0}: {1} seconds".format(self.id(), "%.2f" % (time.time() - self.start_time)))
@staticmethod
def add_corpus(corpus: Corpus):
""" Adds a corpus to the database. """
db.session.add(corpus)
UpdateInfo.query.delete()
ui_cts: UpdateInfo = UpdateInfo(resource_type=ResourceType.cts_data, last_modified_time=datetime.utcnow())
db.session.add(ui_cts)
db.session.commit()
@staticmethod
def clear_folder(folder_path: str):
""" Deletes every file in a folder. """
for f in [x for x in os.listdir(folder_path) if x != ".gitignore"]:
os.remove(os.path.join(folder_path, f))
def test_api_corpus_delete(self):
""" Deletes a single corpus. """
response: Response = self.client.delete(Mocks.corpora[0].uri)
self.assertEqual(response.status_code, 404)
McTestCase.add_corpus(Mocks.corpora[0])
response = self.client.delete(Mocks.corpora[0].uri)
data_json: dict = json.loads(response.get_data())
self.assertEqual(data_json, {"result": True})
Corpus.query.delete()
UpdateInfo.query.delete()
# dirty hack so we can reuse it in other tests
session.make_transient(Mocks.corpora[0])
def test_api_corpus_get(self):
""" Gets information about a single corpus. """
response: Response = self.client.get(Mocks.corpora[0].uri)
self.assertEqual(response.status_code, 404)
McTestCase.add_corpus(Mocks.corpora[0])
response: Response = self.client.get(Mocks.corpora[0].uri)
data_json: dict = json.loads(response.get_data())
actual_corpus: Corpus = Corpus(json_dict=data_json["corpus"])
self.assertEqual(actual_corpus, Mocks.corpora[0])
Corpus.query.delete()
UpdateInfo.query.delete()
# dirty hack so we can reuse it in other tests
session.make_transient(Mocks.corpora[0])
def test_api_corpus_list_get(self):
"""Adds multiple texts to the database and queries them all."""
class MockFilterBy:
def __init__(self, do_raise: bool = False, ui: UpdateInfo = None):
self.do_raise: bool = do_raise
self.ui: UpdateInfo = ui
def first(self):
if self.do_raise:
raise OperationalError("error", [], "")
else:
return self.ui
def expect_result(self: McTestCase, mock: MagicMock, do_raise: bool, lut: str, result: object,
lmt: datetime = datetime.utcnow()):
ui: UpdateInfo = UpdateInfo(resource_type=ResourceType.cts_data, last_modified_time=lmt)
mfb: MockFilterBy = MockFilterBy(do_raise, ui)
mock.query.filter_by.return_value = mfb
response: Response = self.client.get(TestingConfig.SERVER_URI_CORPORA,
query_string=dict(last_update_time=lut))
data_json = json.loads(response.get_data())
if data_json:
data_json = [Corpus(json_dict=x) for x in data_json["corpora"]]
self.assertEqual(data_json, result)
with patch.object(mcserver.app.api.corpusListAPI, "UpdateInfo") as mock_update_info:
expect_result(self, mock_update_info, True, "0", None)
expect_result(self, mock_update_info, False, str(int(datetime.utcnow().timestamp() * 1000)), None,
datetime.fromtimestamp(0))
db.session.add_all(Mocks.corpora)
db.session.commit()
expect_result(self, mock_update_info, False, "0", Mocks.corpora, datetime.fromtimestamp(time.time()))
Corpus.query.delete()
UpdateInfo.query.delete()
# dirty hack so we can reuse it in other tests
session.make_transient(Mocks.corpora[0])
def test_api_corpus_put(self):
""" Changes information about a single corpus. """
response: Response = self.client.put(Mocks.corpora[0].uri)
self.assertEqual(response.status_code, 404)
Loading
Loading full blame...