Newer
Older
"""Unit tests for testing the application functionality."""
import copy
import ntpath
import os
import pickle
import shutil
import sys
import time
import unittest
import uuid
from collections import OrderedDict
from datetime import datetime
from threading import Thread
from typing import Dict, List, Tuple, Type, Any
from unittest import TestLoader
from unittest.mock import patch, MagicMock, mock_open
from zipfile import ZipFile
import MyCapytain
import rapidjson as json
from conllu import TokenList
from flask import Flask
from gensim.models import Word2Vec
Konstantin Schulz
committed
from graphannis.errors import NoSuchCorpus
from lxml import etree
from networkx import MultiDiGraph, Graph
from requests import HTTPError
Konstantin Schulz
committed
from sqlalchemy.exc import OperationalError, InvalidRequestError
from sqlalchemy.orm import session
from werkzeug.wrappers import Response
import mcserver
from mcserver.app import create_app, db, start_updater, full_init, log_exception, get_api_specification, \
init_app_common, create_postgres_database
Konstantin Schulz
committed
from mcserver.app.api.exerciseAPI import map_exercise_data_to_database, get_graph_data
from mcserver.app.api.fileAPI import clean_tmp_folder
from mcserver.app.api.h5pAPI import get_remote_exercise
Konstantin Schulz
committed
from mcserver.app.api.kwicAPI import handle_exercise_data
from mcserver.app.api.vocabularyAPI import check_vocabulary
from mcserver.app.api.zenodoAPI import remove_older_versions
from mcserver.app.models import ResourceType, FileType, ExerciseType, ExerciseData, \
NodeMC, LinkMC, GraphData, Phenomenon, CustomCorpus, AnnisResponse, Solution, DownloadableFile, Language, \
Konstantin Schulz
committed
VocabularyCorpus, TextComplexityMeasure, CitationLevel, FrequencyItem, Choice, XapiStatement, ExerciseMC, \
CorpusMC, make_solution_element_from_salt_id, Sentence, ReferenceableText, Dependency, PartOfSpeech, Citation
Konstantin Schulz
committed
from mcserver.app.services import AnnotationService, CorpusService, FileService, CustomCorpusService, \
DatabaseService, XMLservice, TextService, FrequencyService, ExerciseService, TextComplexityService, NetworkService
from mcserver.config import TestingConfig, Config, DevelopmentConfig
from mcserver.models_auto import Corpus, Exercise, UpdateInfo, LearningResult
from mocks import Mocks, MockResponse, MockW2V, MockQuery, TestHelper
Konstantin Schulz
committed
from openapi.openapi_server.models import VocabularyForm, VocabularyMC, ExerciseForm, \
KwicForm, VectorNetworkForm, MatchingExercise, ExerciseTypePath, H5PForm, TextComplexity, ZenodoRecord, ZenodoForm
class APItestCase(unittest.TestCase):
"""The test suite for the REST API."""
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 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)
TestHelper.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)
TestHelper.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: APItestCase, mock: MagicMock, 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.0)
mock.session.query.return_value = MockQuery(ui)
Konstantin Schulz
committed
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)
Konstantin Schulz
committed
with patch.object(mcserver.app.services.databaseService, "db") as mock_db:
expect_result(self, mock_db, str(int(datetime.utcnow().timestamp() * 1000)), None,
datetime.fromtimestamp(0))
db.session.add_all(Mocks.corpora)
DatabaseService.commit()
expect_result(self, mock_db, "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)
TestHelper.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()
Konstantin Schulz
committed
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)
DatabaseService.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
DatabaseService.commit()
response = Mocks.app_dict[self.class_name].client.get(Config.SERVER_URI_EXERCISE,
query_string=dict(eid=Mocks.exercise.eid))
Konstantin Schulz
committed
ar: AnnisResponse = AnnisResponse.from_dict(json.loads(response.get_data(as_text=True)))
self.assertEqual(len(ar.graph_data.nodes), 52)
db.session.query(Exercise).delete()
session.make_transient(Mocks.exercise)
def test_api_exercise_list_get(self):
""" Retrieves a list of available exercises. """
ui_exercises: UpdateInfo = UpdateInfo.from_dict(resource_type=ResourceType.exercise_list.name,
last_modified_time=1.0, created_time=1.0)
db.session.add(ui_exercises)
DatabaseService.commit()
args: dict = dict(lang="fr", last_update_time=int(time.time()))
response: Response = Mocks.app_dict[self.class_name].client.get(TestingConfig.SERVER_URI_EXERCISE_LIST,
query_string=args)
self.assertEqual(json.loads(response.get_data()), [])
args["last_update_time"] = 0
db.session.add(Mocks.exercise)
DatabaseService.commit()
response = Mocks.app_dict[self.class_name].client.get(TestingConfig.SERVER_URI_EXERCISE_LIST, query_string=args)
exercises: List[MatchingExercise] = []
for exercise_dict in json.loads(response.get_data(as_text=True)):
exercise_dict["search_values"] = json.dumps(exercise_dict["search_values"])
exercise_dict["solutions"] = json.dumps(exercise_dict["solutions"])
exercises.append(MatchingExercise.from_dict(exercise_dict))
self.assertEqual(len(exercises), 1)
args = dict(lang=Language.English.value, vocabulary=VocabularyCorpus.agldt.name, frequency_upper_bound=500)
Loading
Loading full blame...