tests.py

"""Unit tests for testing the application functionality."""
import copy
import logging
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 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.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, SolutionElement, TextComplexityMeasure, FrequencyAnalysis, CitationLevel, FrequencyItem, \
    TextComplexity, Dependency, PartOfSpeech, Choice, XapiStatement, ExerciseMC, CorpusMC
from mcserver.app.services import AnnotationService, CorpusService, FileService, CustomCorpusService, DatabaseService, \
    XMLservice, TextService
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_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)
        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):