<commit_before># -*- coding: utf-8 -*-

"""
Test  dataobjects endpoints
"""

import io
import os
import json
import unittest

from restapi.server import create_app
#from nose2.tools import assert_equals

__author__ = '<NAME> (<EMAIL>)'


class TestDataObjects(unittest.TestCase):

    @classmethod
    def setUpClass(self):
        "set up test fixtures"
        print('### Setting up flask server ###')
        app = create_app()
        app.config['TESTING'] = True
        self.app = app.test_client()

    @classmethod
    def tearDownClass(self):
        "tear down test fixtures"
        print('### Tearing down the flask server ###')

    def test_01_get_verify(self):
        """ Test that the flask server is running and reachable"""

        r = self.app.get('http://localhost:8080/api/verify')
        assert(r.status_code == 200)

    def test_02_post_dataobjects(self):
        """ Test file upload: POST """
        # I need to understand who to reapeat the upload test, since
        # overwrite is not allowed
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         file=(io.BytesIO(b"this is a test"),
                          'test.pdf')))
        assert(r.status_code == 200)  # maybe 201 is more appropriate

    def test_03_post_large_dataobjects(self):
        """ Test large file upload: POST """
        path = os.path.join('/home/irods', 'img.JPG')
        with open(path, "wb") as f:
            f.seek(100000000)  # 100MB file
            f.write(b"\0")
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         file=(open(path, 'rb'), 'img.JPG')))
        os.remove(path)
        assert(r.status_code == 200)  # maybe 201 is more appropriate

    def test_04_get_dataobjects(self):
        """ Test file download: GET """
        deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                 'test.pdf')
        r = self.app.get(deleteURI, data=dict(collection=('/home/guest')))
        assert(r.status_code == 200)
        assert(r.data == b'this is a test')

    def test_05_get_large_dataobjects(self):
        """ Test file download: GET """
        deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                 'img.JPG')
        r = self.app.get(deleteURI, data=dict(collection=('/home/guest')))
        assert(r.status_code == 200)

    def test_06_post_already_existing_dataobjects(self):
        """ Test file upload with already existsing object: POST """
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         file=(io.BytesIO(b"this is a test"),
                          'test.pdf')))
        assert(r.status_code == 400)  # or 409?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors']
        print(error_message)

    def test_07_delete_dataobjects(self):
        """ Test file delete: DELETE """

        # Obatin the list of objects end delete
        r = self.app.get('http://localhost:8080/api/collections')
        assert(r.status_code == 200)
        # is the following correct?
        content = json.loads(r.data.decode('utf-8'))
        # why each element inside data is a list with only 1 element?
        collection = content['Response']['data'][0][0][:-1]
        # I have to remove the zone part.. this should probably be discussed
        collection = '/' + collection.split('/', 2)[2]
        for idx, obj in enumerate(content['Response']['data']):
            if idx == 0:
                continue
            deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                     obj[0])
            r = self.app.delete(deleteURI, data=dict(collection=(collection)))
            assert(r.status_code == 200)

    def test_08_post_dataobjects_in_non_existing_collection(self):
        """ Test file upload in a non existing collection: POST """
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         collection=('/home/wrong/guest'),
                         file=(io.BytesIO(b"this is a test"), 'test.pdf')))
        assert(r.status_code == 400)  # or 409?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors']
        print(error_message)
#       assert('CAT_UNKNOWN_COLLECTION' in error_message)

    def test_09_get_non_exising_dataobjects(self):
        """ Test file download of a non existing object: GET """
        deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                 'test.pdf')
        r = self.app.get(deleteURI, data=dict(collection=('/home/guest')))
        assert(r.status_code == 400)  # or 404?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors']
        print(error_message)
#       assert('USER_INPUT_PATH_ERR' in error_message)

    def test_10_get_dataobjects_in_non_exising_collection(self):
        """ Test file download in a non existing collection: GET """
        delURI = os.path.join('http://localhost:8080/api/dataobjects',
                              'test.pdf')
        r = self.app.get(delURI, data=dict(collection=('/home/wrong/guest')))
        assert(r.status_code == 400)  # or 404?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors']
        print(error_message)
#       assert('USER_INPUT_PATH_ERR' in error_message)
<commit_msg>Better use of asserts<commit_after># -*- coding: utf-8 -*-

"""
Test  dataobjects endpoints
"""

import io
import os
import json
import unittest
from restapi.server import create_app


__author__ = '<NAME> (<EMAIL>)'


class TestDataObjects(unittest.TestCase):

    @classmethod
    def setUpClass(self):
        "set up test fixtures"
        print('### Setting up flask server ###')
        app = create_app()
        app.config['TESTING'] = True
        self.app = app.test_client()

    @classmethod
    def tearDownClass(self):
        "tear down test fixtures"
        print('### Tearing down the flask server ###')

    def test_01_get_verify(self):
        """ Test that the flask server is running and reachable"""

        r = self.app.get('http://localhost:8080/api/verify')
        self.assertEqual(r.status_code, 200)

    def test_02_post_dataobjects(self):
        """ Test file upload: POST """
        # I need to understand who to reapeat the upload test, since
        # overwrite is not allowed
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         file=(io.BytesIO(b"this is a test"),
                          'test.pdf')))
        self.assertEqual(r.status_code, 200)  # maybe 201 is more appropriate

    def test_03_post_large_dataobjects(self):
        """ Test large file upload: POST """
        path = os.path.join('/home/irods', 'img.JPG')
        with open(path, "wb") as f:
            f.seek(100000000)  # 100MB file
            f.write(b"\0")
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         file=(open(path, 'rb'), 'img.JPG')))
        os.remove(path)
        self.assertEqual(r.status_code, 200)  # maybe 201 is more appropriate

    def test_04_get_dataobjects(self):
        """ Test file download: GET """
        deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                 'test.pdf')
        r = self.app.get(deleteURI, data=dict(collection=('/home/guest')))
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.data, b'this is a test')

    def test_05_get_large_dataobjects(self):
        """ Test file download: GET """
        deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                 'img.JPG')
        r = self.app.get(deleteURI, data=dict(collection=('/home/guest')))
        self.assertEqual(r.status_code, 200)

    def test_06_post_already_existing_dataobjects(self):
        """ Test file upload with already existsing object: POST """
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         file=(io.BytesIO(b"this is a test"),
                          'test.pdf')))
        self.assertEqual(r.status_code, 400)  # or 409?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors']
        print(error_message)

    def test_07_delete_dataobjects(self):
        """ Test file delete: DELETE """

        # Obatin the list of objects end delete
        r = self.app.get('http://localhost:8080/api/collections')
        self.assertEqual(r.status_code, 200)
        # is the following correct?
        content = json.loads(r.data.decode('utf-8'))
        # why each element inside data is a list with only 1 element?
        collection = content['Response']['data'][0][0][:-1]
        # I have to remove the zone part.. this should probably be discussed
        collection = '/' + collection.split('/', 2)[2]
        for idx, obj in enumerate(content['Response']['data']):
            if idx == 0:
                continue
            deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                     obj[0])
            r = self.app.delete(deleteURI, data=dict(collection=(collection)))
            self.assertEqual(r.status_code, 200)

    def test_08_post_dataobjects_in_non_existing_collection(self):
        """ Test file upload in a non existing collection: POST """
        r = self.app.post('http://localhost:8080/api/dataobjects', data=dict(
                         collection=('/home/wrong/guest'),
                         file=(io.BytesIO(b"this is a test"), 'test.pdf')))
        self.assertEqual(r.status_code, 400)  # or 409?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors'][0]['iRODS']
        self.assertIn('CAT_UNKNOWN_COLLECTION', error_message)

    def test_09_get_non_exising_dataobjects(self):
        """ Test file download of a non existing object: GET """
        deleteURI = os.path.join('http://localhost:8080/api/dataobjects',
                                 'test.pdf')
        r = self.app.get(deleteURI, data=dict(collection=('/home/guest')))
        self.assertEqual(r.status_code, 400)  # or 404?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors'][0]['iRODS']
        self.assertIn('USER_INPUT_PATH_ERR', error_message)

    def test_10_get_dataobjects_in_non_exising_collection(self):
        """ Test file download in a non existing collection: GET """
        delURI = os.path.join('http://localhost:8080/api/dataobjects',
                              'test.pdf')
        r = self.app.get(delURI, data=dict(collection=('/home/wrong/guest')))
        self.assertEqual(r.status_code, 400)  # or 404?
        content = json.loads(r.data.decode('utf-8'))
        error_message = content['Response']['errors'][0]['iRODS']
        self.assertIn('USER_INPUT_PATH_ERR', error_message)

{'fraction_non_alphanumeric': 0.13790948852985807, 'fraction_numerical': 0.02570739980362403, 'mean_word_length': 2.3696240601503757, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1326650', 'n_tokens_mistral': 3512, 'n_tokens_neox': 3240, 'n_words': 715}