tests.py

"""Unit tests for testing the application functionality."""
# TODO: TEST BLINKER, GUNICORN, PSYCOPG2-BINARY AND COVERAGE IMPORTS
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)