<commit_before>"""Test for checking if the response format is proper. Run test_crud before running this."""
import unittest
import random
import string
import json
import re
import uuid
from hydrus.app_factory import app_factory
from hydrus.utils import set_session, set_doc, set_api_name
from hydrus.data import doc_parse, crud
from hydra_python_core import doc_maker
from hydrus.samples import doc_writer_sample
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from hydrus.data.db_models import Base


def gen_dummy_object(class_, doc):
    """Create a dummy object based on the definitions in the API Doc."""
    object_ = {
        "@type": class_
    }
    if class_ in doc.parsed_classes:
        for prop in doc.parsed_classes[class_]["class"].supportedProperty:
            if "vocab:" in prop.prop:
                prop_class = prop.prop.replace("vocab:", "")
                object_[prop.title] = gen_dummy_object(prop_class, doc)
            else:
                object_[prop.title] = ''.join(random.choice(
                    string.ascii_uppercase + string.digits) for _ in range(6))
        return object_


class ViewsTestCase(unittest.TestCase):
    """Test Class for the app."""

    @classmethod
    def setUpClass(self):
        """Database setup before the tests."""
        print("Creating a temporary database...")
        engine = create_engine('sqlite:///:memory:')
        Base.metadata.create_all(engine)
        session = scoped_session(sessionmaker(bind=engine))

        self.session = session
        self.API_NAME = "demoapi"
        self.HYDRUS_SERVER_URL = "http://hydrus.com/"

        self.app = app_factory(self.API_NAME)
        print("going for create doc")

        self.doc = doc_maker.create_doc(
            doc_writer_sample.api_doc.generate(),
            self.HYDRUS_SERVER_URL,
            self.API_NAME)
        test_classes = doc_parse.get_classes(self.doc.generate())
        test_properties = doc_parse.get_all_properties(test_classes)
        doc_parse.insert_classes(test_classes, self.session)
        doc_parse.insert_properties(test_properties, self.session)

        print("Classes and properties added successfully.")

        print("Setting up Hydrus utilities... ")
        self.api_name_util = set_api_name(self.app, self.API_NAME)
        self.session_util = set_session(self.app, self.session)
        self.doc_util = set_doc(self.app, self.doc)
        self.client = self.app.test_client()

        print("Creating utilities context... ")
        self.api_name_util.__enter__()
        self.session_util.__enter__()
        self.doc_util.__enter__()
        self.client.__enter__()

        print("Setup done, running tests...")

    @classmethod
    def tearDownClass(self):
        """Tear down temporary database and exit utilities"""
        self.client.__exit__(None, None, None)
        self.doc_util.__exit__(None, None, None)
        self.session_util.__exit__(None, None, None)
        self.api_name_util.__exit__(None, None, None)
        self.session.close()

    def setUp(self):
        for class_ in self.doc.parsed_classes:
            if class_ not in self.doc.collections:
                dummy_obj = gen_dummy_object(class_, self.doc)
                crud.insert(
                    dummy_obj,
                    id_=str(
                        uuid.uuid4()),
                    session=self.session)

    def test_Index(self):
        """Test for the index."""
        response_get = self.client.get("/{}".format(self.API_NAME))
        endpoints = json.loads(response_get.data.decode('utf-8'))
        response_post = self.client.post(
            "/{}".format(self.API_NAME), data=dict(foo="bar"))
        response_put = self.client.put(
            "/{}".format(self.API_NAME), data=dict(foo="bar"))
        response_delete = self.client.delete("/{}".format(self.API_NAME))
        assert "@context" in endpoints
        assert endpoints["@id"] == "/{}".format(self.API_NAME)
        assert endpoints["@type"] == "EntryPoint"
        assert response_get.status_code == 200
        assert response_post.status_code == 405
        assert response_put.status_code == 405
        assert response_delete.status_code == 405

    def test_EntryPoint_context(self):
        """Test for the EntryPoint context."""
        response_get = self.client.get(
            "/{}/contexts/EntryPoint.jsonld".format(self.API_NAME))
        response_get_data = json.loads(response_get.data.decode('utf-8'))
        response_post = self.client.post(
            "/{}/contexts/EntryPoint.jsonld".format(self.API_NAME), data={})
        response_delete = self.client.delete(
            "/{}/contexts/EntryPoint.jsonld".format(self.API_NAME))
        assert response_get.status_code == 200
        assert "@context" in response_get_data
        assert response_post.status_code == 405
        assert response_delete.status_code == 405

    def test_Vocab(self):
        """Test the vocab."""
        response_get = self.client.get("/{}/vocab#".format(self.API_NAME))
        response_get_data = json.loads(response_get.data.decode('utf-8'))

        assert "@context" in response_get_data
        assert response_get_data["@type"] == "ApiDocumentation"
        assert response_get_data["@id"] == "{}{}/vocab".format(
            self.HYDRUS_SERVER_URL, self.API_NAME)
        assert response_get.status_code == 200

        response_delete = self.client.delete(
            "/{}/vocab#".format(self.API_NAME))
        assert response_delete.status_code == 405

        response_put = self.client.put(
            "/{}/vocab#".format(self.API_NAME), data=json.dumps(dict(foo='bar')))
        assert response_put.status_code == 405

        response_post = self.client.post(
            "/{}/vocab#".format(self.API_NAME), data=json.dumps(dict(foo='bar')))
        assert response_post.status_code == 405

    def test_Collections_GET(self):
        """Test GET on collection endpoints."""
        index = self.client.get("/{}".format(self.API_NAME))
        assert index.status_code == 200
        endpoints = json.loads(index.data.decode('utf-8'))
        for endpoint in endpoints:
            collection_name = "/".join(endpoints[endpoint].split(
                "/{}/".format(self.API_NAME))[1:])
            if collection_name in self.doc.collections:
                response_get = self.client.get(endpoints[endpoint])
                # pdb.set_trace()
                assert response_get.status_code == 200
                response_get_data = json.loads(
                    response_get.data.decode('utf-8'))
                assert "@context" in response_get_data
                assert "@id" in response_get_data
                assert "@type" in response_get_data
                assert "members" in response_get_data

    def test_Collections_PUT(self):
        """Test insert data to the collection."""
        index = self.client.get("/{}".format(self.API_NAME))<commit_msg>Add tests for pagination feature<commit_after>"""Test for checking if the response format is proper. Run test_crud before running this."""
import unittest
import random
import string
import json
import re
import uuid
from hydrus.app_factory import app_factory
from hydrus.utils import set_session, set_doc, set_api_name, set_page_size
from hydrus.data import doc_parse, crud
from hydra_python_core import doc_maker
from hydrus.samples import doc_writer_sample
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from hydrus.data.db_models import Base


def gen_dummy_object(class_, doc):
    """Create a dummy object based on the definitions in the API Doc."""
    object_ = {
        "@type": class_
    }
    if class_ in doc.parsed_classes:
        for prop in doc.parsed_classes[class_]["class"].supportedProperty:
            if "vocab:" in prop.prop:
                prop_class = prop.prop.replace("vocab:", "")
                object_[prop.title] = gen_dummy_object(prop_class, doc)
            else:
                object_[prop.title] = ''.join(random.choice(
                    string.ascii_uppercase + string.digits) for _ in range(6))
        return object_


class ViewsTestCase(unittest.TestCase):
    """Test Class for the app."""

    @classmethod
    def setUpClass(self):
        """Database setup before the tests."""
        print("Creating a temporary database...")
        engine = create_engine('sqlite:///:memory:')
        Base.metadata.create_all(engine)
        session = scoped_session(sessionmaker(bind=engine))

        self.session = session
        self.API_NAME = "demoapi"
        self.page_size = 1
        self.HYDRUS_SERVER_URL = "http://hydrus.com/"

        self.app = app_factory(self.API_NAME)
        print("going for create doc")

        self.doc = doc_maker.create_doc(
            doc_writer_sample.api_doc.generate(),
            self.HYDRUS_SERVER_URL,
            self.API_NAME)
        test_classes = doc_parse.get_classes(self.doc.generate())
        test_properties = doc_parse.get_all_properties(test_classes)
        doc_parse.insert_classes(test_classes, self.session)
        doc_parse.insert_properties(test_properties, self.session)

        print("Classes and properties added successfully.")

        print("Setting up hydrus utilities... ")
        self.api_name_util = set_api_name(self.app, self.API_NAME)
        self.session_util = set_session(self.app, self.session)
        self.doc_util = set_doc(self.app, self.doc)
        self.page_size_util = set_page_size(self.app, self.page_size)
        self.client = self.app.test_client()

        print("Creating utilities context... ")
        self.api_name_util.__enter__()
        self.session_util.__enter__()
        self.doc_util.__enter__()
        self.client.__enter__()

        print("Setup done, running tests...")

    @classmethod
    def tearDownClass(self):
        """Tear down temporary database and exit utilities"""
        self.client.__exit__(None, None, None)
        self.doc_util.__exit__(None, None, None)
        self.session_util.__exit__(None, None, None)
        self.api_name_util.__exit__(None, None, None)
        self.session.close()

    def setUp(self):
        for class_ in self.doc.parsed_classes:
            dummy_obj = gen_dummy_object(class_, self.doc)
            crud.insert(
                dummy_obj,
                id_=str(
                    uuid.uuid4()),
                session=self.session)
            # If it's a collection class then add an extra object so we can test pagination thoroughly.
            if class_ in self.doc.collections:
                crud.insert(
                    dummy_obj,
                    id_=str(
                        uuid.uuid4()),
                    session=self.session)

    def test_Index(self):
        """Test for the index."""
        response_get = self.client.get("/{}".format(self.API_NAME))
        endpoints = json.loads(response_get.data.decode('utf-8'))
        response_post = self.client.post(
            "/{}".format(self.API_NAME), data=dict(foo="bar"))
        response_put = self.client.put(
            "/{}".format(self.API_NAME), data=dict(foo="bar"))
        response_delete = self.client.delete("/{}".format(self.API_NAME))
        assert "@context" in endpoints
        assert endpoints["@id"] == "/{}".format(self.API_NAME)
        assert endpoints["@type"] == "EntryPoint"
        assert response_get.status_code == 200
        assert response_post.status_code == 405
        assert response_put.status_code == 405
        assert response_delete.status_code == 405

    def test_EntryPoint_context(self):
        """Test for the EntryPoint context."""
        response_get = self.client.get(
            "/{}/contexts/EntryPoint.jsonld".format(self.API_NAME))
        response_get_data = json.loads(response_get.data.decode('utf-8'))
        response_post = self.client.post(
            "/{}/contexts/EntryPoint.jsonld".format(self.API_NAME), data={})
        response_delete = self.client.delete(
            "/{}/contexts/EntryPoint.jsonld".format(self.API_NAME))
        assert response_get.status_code == 200
        assert "@context" in response_get_data
        assert response_post.status_code == 405
        assert response_delete.status_code == 405

    def test_Vocab(self):
        """Test the vocab."""
        response_get = self.client.get("/{}/vocab#".format(self.API_NAME))
        response_get_data = json.loads(response_get.data.decode('utf-8'))

        assert "@context" in response_get_data
        assert response_get_data["@type"] == "ApiDocumentation"
        assert response_get_data["@id"] == "{}{}/vocab".format(
            self.HYDRUS_SERVER_URL, self.API_NAME)
        assert response_get.status_code == 200

        response_delete = self.client.delete(
            "/{}/vocab#".format(self.API_NAME))
        assert response_delete.status_code == 405

        response_put = self.client.put(
            "/{}/vocab#".format(self.API_NAME), data=json.dumps(dict(foo='bar')))
        assert response_put.status_code == 405

        response_post = self.client.post(
            "/{}/vocab#".format(self.API_NAME), data=json.dumps(dict(foo='bar')))
        assert response_post.status_code == 405

    def test_Collections_GET(self):
        """Test GET on collection endpoints."""
        index = self.client.get("/{}".format(self.API_NAME))
        assert index.status_code == 200
        endpoints = json.loads(index.data.decode('utf-8'))
        for endpoint in endpoints:
            collection_name = "/".join(endpoints[endpoint].split(
                "/{}/".format(self.API_NAME))[1:])
            if collection_name in self.doc.collections:
                response_get = self.client.get(endpoints[endpoint])
                # pdb.set_trace()
                assert response_get.status_code == 200
                response_get_data = json.loads(
                    response_get.data.decode('utf-8'))
                assert "@context" in response_get_data
                assert "@id" in response_get_data
                assert "@type" in response_get_data
                assert "members" in response_get_data

    def test_pagination(self):
        """Test basic pagination"""
        index = self.client.get("/{}".format(self.API_NAME))
        assert index.status_code == 200
        endpoints = json.loads(index.data.decode('utf-8'))
        for endpoint in endpoints:
            collection_name = "/".join(endpoints[endpoint].split(
                "/{}/".format(self.API_NAME))[1:])
            if collection_name in self.doc.collections:
                response_get = self.client.get(endpoints[endpoint])
                assert response_get.status_code == 200
                response_get_data = json.loads(
                    response_get.data.decode('utf-8'))
                assert "view" in response_get_data
                assert "first" in response_get_data["view"]
                assert "last" in response_get_data["view"]
                if "next" in response_get_data["view"]:
                    response_next = self.client.get(response_get_data["view"]["next"])
                    assert response_next.status_code == 200
                    response_next_data = json.loads(
                        response_next.data.decode('utf-8'))
                    assert "previous" in response_next_data["view"]
                break

    def test_Collections_PUT(self):
        """Test insert data to the collection."""
        index = self.client.get("/{}".format(self.API_NAME))
{'fraction_non_alphanumeric': 0.1171609762347063, 'fraction_numerical': 0.006982256101466915, 'mean_word_length': 2.4387665198237887, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '84637', 'n_tokens_mistral': 4618, 'n_tokens_neox': 4410, 'n_words': 903}