Newer
Older
"""Unit tests for testing the application functionality."""
import copy
import ntpath
import os
import uuid
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, Any
from conllu import TokenList
from flask import Flask
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.wrappers import Response
import csm
import mcserver
from csm 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.models import ResourceType, FileType, ExerciseType, ExerciseData, \
NodeMC, LinkMC, GraphData, Phenomenon, CustomCorpus, AnnisResponse, Solution, DownloadableFile, Language, \
VocabularyCorpus, TextComplexityMeasure, CitationLevel, FrequencyItem, TextComplexity, Dependency, PartOfSpeech, \
Choice, XapiStatement, ExerciseMC, CorpusMC, make_solution_element_from_salt_id
from mcserver.app.services import AnnotationService, CorpusService, FileService, CustomCorpusService, DatabaseService, \
XMLservice, TextService, FrequencyService
from mcserver.config import TestingConfig, Config
from mcserver.models_auto import Corpus, Exercise, UpdateInfo, LearningResult
from mocks import Mocks, MockResponse, MockW2V, MockQuery, TestHelper
class McTestCase(unittest.TestCase):
"""The test suite for the main application."""
def mocked_requests_get(*args, **kwargs):
if TestingConfig.SIMULATE_HTTP_ERROR:
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))
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)
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)
self.class_name: str = str(self.__class__)
TestHelper.update_flask_app(self.class_name, create_app)
def tearDown(self):
"""Finishes testing by removing the traces."""
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)
db.session.query(UpdateInfo).delete()
ui_cts: UpdateInfo = UpdateInfo.from_dict(resource_type=ResourceType.cts_data.name,
last_modified_time=datetime.utcnow().timestamp(), created_time=1)
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_bad_request(self):
"""Returns validation errors as JSON."""
response: Response = Mocks.app_dict[self.class_name].client.get(Config.SERVER_URI_CORPORA)
self.assertEqual(response.status_code, 400)
def test_api_corpus_delete(self):
""" Deletes a single corpus. """
db.session.query(Corpus).delete()
response: Response = Mocks.app_dict[self.class_name].client.delete(
f"{Config.SERVER_URI_CORPORA}/1")
self.assertEqual(response.status_code, 404)
McTestCase.add_corpus(Mocks.corpora[0])
response = Mocks.app_dict[self.class_name].client.delete(f"{Config.SERVER_URI_CORPORA}/{Mocks.corpora[0].cid}")
data_json: dict = json.loads(response.get_data())
self.assertEqual(data_json, True)
db.session.query(Corpus).delete()
db.session.query(UpdateInfo).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 = Mocks.app_dict[self.class_name].client.get(
f"{Config.SERVER_URI_CORPORA}/{Mocks.corpora[0].cid}")
self.assertEqual(response.status_code, 404)
McTestCase.add_corpus(Mocks.corpora[0])
response: Response = Mocks.app_dict[self.class_name].client.get(
f"{Config.SERVER_URI_CORPORA}/{Mocks.corpora[0].cid}")
data_json: dict = json.loads(response.get_data())
old_dict: dict = Mocks.corpora[0].to_dict()
self.assertEqual(data_json, old_dict)
db.session.query(Corpus).delete()
db.session.query(UpdateInfo).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."""
def expect_result(self: McTestCase, mock: MagicMock, do_raise: bool, lut: str, result: Any,
lmt: datetime = datetime.utcnow()):
ui: UpdateInfo = UpdateInfo.from_dict(resource_type=ResourceType.cts_data.name,
last_modified_time=lmt.timestamp(), created_time=1)
mock.session.query.return_value = MockQuery(do_raise, ui)
response: Response = Mocks.app_dict[self.class_name].client.get(TestingConfig.SERVER_URI_CORPORA,
query_string=dict(last_update_time=lut))
data_json = json.loads(response.get_data())
if data_json:
result = [x.to_dict() for x in result]
self.assertEqual(data_json, result)
with patch.object(mcserver.app.api.corpusListAPI, "db") as mock_db:
expect_result(self, mock_db, True, "0", None)
expect_result(self, mock_db, False, str(int(datetime.utcnow().timestamp() * 1000)), None,
datetime.fromtimestamp(0))
db.session.add_all(Mocks.corpora)
db.session.commit()
expect_result(self, mock_db, False, "0", Mocks.corpora, datetime.fromtimestamp(time.time()))
db.session.query(Corpus).delete()
db.session.query(UpdateInfo).delete()
# dirty hack so we can reuse it in other tests
session.make_transient(Mocks.corpora[0])
def test_api_corpus_patch(self):
""" Changes information about a single corpus. """
response: Response = Mocks.app_dict[self.class_name].client.patch(
f"{Config.SERVER_URI_CORPORA}/{Mocks.corpora[0].cid}")
self.assertEqual(response.status_code, 404)
McTestCase.add_corpus(Mocks.corpora[0])
old_title: str = Mocks.corpora[0].title
new_title: str = "new_title"
response: Response = Mocks.app_dict[self.class_name].client.patch(
f"{Config.SERVER_URI_CORPORA}/{Mocks.corpora[0].cid}", data=dict(title=new_title))
data_json: dict = json.loads(response.get_data())
old_dict: dict = Mocks.corpora[0].to_dict()
self.assertEqual(data_json["title"], old_dict["title"])
Mocks.corpora[0].title = old_title
db.session.query(Corpus).delete()
db.session.query(UpdateInfo).delete()
# dirty hack so we can reuse it in other tests
session.make_transient(Mocks.corpora[0])
def test_api_exercise_get(self):
""" Retrieves an existing exercise by its exercise ID. """
db.session.query(Exercise).delete()
response: Response = Mocks.app_dict[self.class_name].client.get(Config.SERVER_URI_EXERCISE,
query_string=dict(eid=""))
self.assertEqual(response.status_code, 404)
old_urn: str = Mocks.exercise.urn
Mocks.exercise.urn = ""
db.session.add(Mocks.exercise)
db.session.commit()
ar: AnnisResponse = AnnisResponse(solutions=[], graph_data=GraphData(links=[], nodes=[]))
with patch.object(CorpusService, "get_corpus", side_effect=[ar, Mocks.annis_response]):
response = Mocks.app_dict[self.class_name].client.get(Config.SERVER_URI_EXERCISE,
query_string=dict(eid=Mocks.exercise.eid))
self.assertEqual(response.status_code, 404)
Mocks.exercise.urn = old_urn
db.session.commit()
response = Mocks.app_dict[self.class_name].client.get(Config.SERVER_URI_EXERCISE,
query_string=dict(eid=Mocks.exercise.eid))
graph_dict: dict = json.loads(response.get_data(as_text=True))
ar: AnnisResponse = AnnisResponse.from_dict(graph_dict)
self.assertEqual(len(ar.graph_data.nodes), 52)
Loading
Loading full blame...