biosample.tests.test_async_biosample.AsyncBioSamplesTestCase.test_request_with_html() - Code Metrics - Inspection of ":bookmark: InjectTool release v0.9.5" - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#97)

by Paolo

created 2020-06-04 17:49 UTC

AsyncBioSamplesTestCase.test_request_with_html() A

↳ Parent: biosample.tests.test_async_biosample

Complexity

Conditions

Size

Total Lines	20
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	15
dl	0
loc	20
rs	9.65
c	0
b	0
f	0
cc	2
nop	1

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Tue Jan 21 10:54:09 2020

@author: Paolo Cozzi <[email protected]>
"""

import os
import json
import time
import types
import asynctest

from aioresponses import aioresponses
from aiohttp.client_exceptions import ServerDisconnectedError
from django.test import TestCase
from unittest.mock import patch, Mock

from common.constants import BIOSAMPLE_URL, SUBMITTED
from uid.models import Animal as UIDAnimal, Sample as UIDSample

from ..tasks.cleanup import check_samples, get_orphan_samples, PAGE_SIZE
from ..models import OrphanSample, ManagedTeam

from .common import generate_token, BioSamplesMixin

# get my path
dir_path = os.path.dirname(os.path.realpath(__file__))

# define data path
DATA_PATH = os.path.join(dir_path, "data")


with open(os.path.join(DATA_PATH, "page_0.json")) as handle:
    page0 = handle.read()

with open(os.path.join(DATA_PATH, "page_1.json")) as handle:
    page1 = handle.read()

with open(os.path.join(DATA_PATH, "issue_page1.json")) as handle:
    issue_page1 = handle.read()


class AsyncBioSamplesTestCase(asynctest.TestCase, TestCase):

    fixtures = [
        'biosample/managedteam',
        'uid/animal',
        'uid/dictbreed',
        'uid/dictcountry',
        'uid/dictrole',
        'uid/dictsex',
        'uid/dictspecie',
        'uid/dictstage',
        'uid/dictuberon',
        'uid/ontology',
        'uid/organization',
        'uid/publication',
        'uid/sample',
        'uid/submission',
        'uid/user'
    ]

    @classmethod
    def setUpClass(cls):
        # calling my base class setup
        super().setUpClass()

        cls.mock_auth_patcher = patch('pyUSIrest.auth.requests.get')
        cls.mock_auth = cls.mock_auth_patcher.start()

    @classmethod
    def tearDownClass(cls):
        cls.mock_auth_patcher.stop()

        # calling base method
        super().tearDownClass()

    def setUp(self):
        # calling my base setup
        super().setUp()

        # well, updating data and set two biosample ids. Those are not
        # orphans
        animal = UIDAnimal.objects.get(pk=1)
        animal.biosample_id = "SAMEA6376980"
        animal.save()

        sample = UIDSample.objects.get(pk=1)
        sample.biosample_id = "SAMEA6376982"
        sample.save()

        # generate tocken
        self.mock_auth.return_value = Mock()
        self.mock_auth.return_value.text = generate_token()
        self.mock_auth.return_value.status_code = 200

    async def test_request(self) -> None:
        with aioresponses() as mocked:
            mocked.get(
                '{url}?filter=attr:project:IMAGE&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                body=page0)
            mocked.get(
                '{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                body=page1)

            await check_samples()

            # get accessions
            reference = ['SAMEA6376991', 'SAMEA6376992']

            self.assertEqual(OrphanSample.objects.count(), 2)

            # check objects into UID
            for accession in reference:
                qs = OrphanSample.objects.filter(biosample_id=accession)
                self.assertTrue(qs.exists())

    async def test_request_with_issues(self) -> None:
        """Test a temporary issue with BioSamples reply"""

        with aioresponses() as mocked:
            mocked.get(
                '{url}?filter=attr:project:IMAGE&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                body=page0)
            mocked.get(
                '{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                body=issue_page1)

            await check_samples()

        # no objects where tracked since issue in response
        self.assertEqual(OrphanSample.objects.count(), 0)

    async def test_request_with_html(self) -> None:
        """Test a not JSON response (HTML)"""

        with aioresponses() as mocked:
            mocked.get(
                '{url}?filter=attr:project:IMAGE&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                body=page0)
            mocked.get(
                '{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                headers={'Content-type': 'text/html'},
                body="<html>Not a JSON</html>")

            await check_samples()

        # no objects where tracked since issue in response
        self.assertEqual(OrphanSample.objects.count(), 0)

    async def test_biosamples_down(self) -> None:
        """Test a not JSON response (HTML): BioSamples down"""

        with aioresponses() as mocked:
            mocked.get(
                '{url}?filter=attr:project:IMAGE&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                headers={'Content-type': 'text/html'},
                body="<html>Not a JSON</html>")
            mocked.get(
                '{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                status=200,
                headers={'Content-type': 'text/html'},
                body="<html>Not a JSON</html>")

            with self.assertRaises(ConnectionError):
                await check_samples()

        # no objects where tracked since issue in response
        self.assertEqual(OrphanSample.objects.count(), 0)

    async def test_server_lost(self) -> None:
        """Test server disconnect error"""

        with aioresponses() as mocked:
            mocked.get(
                '{url}?filter=attr:project:IMAGE&size={size}'.format(
                    url=BIOSAMPLE_URL, size=PAGE_SIZE),
                exception=ServerDisconnectedError()
            )

            with self.assertRaises(ConnectionError):
                await check_samples()

        # no objects where tracked since issue in response
        self.assertEqual(OrphanSample.objects.count(), 0)


class PurgeOrphanSampleTestCase(BioSamplesMixin, TestCase):
    fixtures = [
        'biosample/managedteam',
        'biosample/orphansample',
        'uid/dictspecie',
    ]

    def test_purge_orphan_samples(self):
        """Test biosample data conversion"""

        with open(os.path.join(DATA_PATH, "SAMEA6376982.json")) as handle:
            data = json.load(handle)

        self.mock_get.return_value = Mock()
        self.mock_get.return_value.json.return_value = data
        self.mock_get.return_value.status_code = 200

        # call my method
        samples = get_orphan_samples()

        # teams is now a generator
        self.assertIsInstance(samples, types.GeneratorType)
        samples = list(samples)

        self.assertEqual(len(samples), 2)

        sample = samples[0]
        self.assertIsInstance(sample, dict)

        sample = samples[1]
        self.assertIsInstance(sample, dict)

        # read the team from data
        team = sample['team']
        self.assertIsInstance(team, ManagedTeam)

    def test_purge_orphan_samples_not_ready(self):
        """Test not ready orphan samples"""

        # Simulate a different status
        OrphanSample.objects.update(status=SUBMITTED)
        orphan_count = sum(1 for orphan in get_orphan_samples())

        self.assertEqual(orphan_count, 0)

    def test_purge_orphan_samples_ignore(self):
        """Test ignored orphan samples"""

        # Ignoring samples gives no object
        OrphanSample.objects.update(ignore=True)
        orphan_count = sum(1 for orphan in get_orphan_samples())

        self.assertEqual(orphan_count, 0)

    def test_purge_orphan_samples_removed(self):
        """Test removed orphan samples"""

        # Ignoring samples gives no object
        OrphanSample.objects.update(removed=True)
        orphan_count = sum(1 for orphan in get_orphan_samples())

        self.assertEqual(orphan_count, 0)

    def test_purge_orphan_samples_with_limit(self):
        """Test get orphan samples with limits"""

        orphan_count = sum(1 for orphan in get_orphan_samples(limit=1))
        self.assertEqual(orphan_count, 1)

    def test_purge_orphan_private(self):
        """Test no access to a BioSamples id (already removed?)"""

        data = {
            'timestamp': int(time.time() * 1000),
            'status': 403,
            'error': 'Forbidden',
            'exception': (
                'uk.ac.ebi.biosamples.service.'
                'BioSamplesAapService$SampleNotAccessibleException'),
            'message': (
                'This sample is private and not available for browsing. '
                'If you think this is an error and/or you should have access '
                'please contact the BioSamples Helpdesk at biosamples@'
                'ebi.ac.uk'),
            'path': '/biosamples/samples/SAMEA6376982'
        }

        # override mock object
        self.mock_get.return_value = Mock()
        self.mock_get.return_value.json.return_value = data
        self.mock_get.return_value.status_code = 403

        orphan_count = sum(1 for orphan in get_orphan_samples(limit=1))
        self.assertEqual(orphan_count, 0)


1			#!/usr/bin/env python3
2			# -- coding: utf-8 --
3			"""
4			Created on Tue Jan 21 10:54:09 2020
5
6			@author: Paolo Cozzi <[email protected]>
7			"""
8
9			import os
10			import json
11			import time
12			import types
13			import asynctest
14
15			from aioresponses import aioresponses
16			from aiohttp.client_exceptions import ServerDisconnectedError
17			from django.test import TestCase
18			from unittest.mock import patch, Mock
19
20			from common.constants import BIOSAMPLE_URL, SUBMITTED
21			from uid.models import Animal as UIDAnimal, Sample as UIDSample
22
23			from ..tasks.cleanup import check_samples, get_orphan_samples, PAGE_SIZE
24			from ..models import OrphanSample, ManagedTeam
25
26			from .common import generate_token, BioSamplesMixin
27
28			# get my path
29			dir_path = os.path.dirname(os.path.realpath(__file__))
30
31			# define data path
32			DATA_PATH = os.path.join(dir_path, "data")
33
34
35			with open(os.path.join(DATA_PATH, "page_0.json")) as handle:
36			page0 = handle.read()
37
38			with open(os.path.join(DATA_PATH, "page_1.json")) as handle:
39			page1 = handle.read()
40
41			with open(os.path.join(DATA_PATH, "issue_page1.json")) as handle:
42			issue_page1 = handle.read()
43
44
45			class AsyncBioSamplesTestCase(asynctest.TestCase, TestCase):
46
47			fixtures = [
48			'biosample/managedteam',
49			'uid/animal',
50			'uid/dictbreed',
51			'uid/dictcountry',
52			'uid/dictrole',
53			'uid/dictsex',
54			'uid/dictspecie',
55			'uid/dictstage',
56			'uid/dictuberon',
57			'uid/ontology',
58			'uid/organization',
59			'uid/publication',
60			'uid/sample',
61			'uid/submission',
62			'uid/user'
63			]
64
65			@classmethod
66			def setUpClass(cls):
67			# calling my base class setup
68			super().setUpClass()
69
70			cls.mock_auth_patcher = patch('pyUSIrest.auth.requests.get')
71			cls.mock_auth = cls.mock_auth_patcher.start()
72
73			@classmethod
74			def tearDownClass(cls):
75			cls.mock_auth_patcher.stop()
76
77			# calling base method
78			super().tearDownClass()
79
80			def setUp(self):
81			# calling my base setup
82			super().setUp()
83
84			# well, updating data and set two biosample ids. Those are not
85			# orphans
86			animal = UIDAnimal.objects.get(pk=1)
87			animal.biosample_id = "SAMEA6376980"
88			animal.save()
89
90			sample = UIDSample.objects.get(pk=1)
91			sample.biosample_id = "SAMEA6376982"
92			sample.save()
93
94			# generate tocken
95			self.mock_auth.return_value = Mock()
96			self.mock_auth.return_value.text = generate_token()
97			self.mock_auth.return_value.status_code = 200
98
99			async def test_request(self) -> None:
100			with aioresponses() as mocked:
101			mocked.get(
102			'{url}?filter=attr:project:IMAGE&size={size}'.format(
103			url=BIOSAMPLE_URL, size=PAGE_SIZE),
104			status=200,
105			body=page0)
106			mocked.get(
107			'{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
108			url=BIOSAMPLE_URL, size=PAGE_SIZE),
109			status=200,
110			body=page1)
111
112			await check_samples()
113
114			# get accessions
115			reference = ['SAMEA6376991', 'SAMEA6376992']
116
117			self.assertEqual(OrphanSample.objects.count(), 2)
118
119			# check objects into UID
120			for accession in reference:
121			qs = OrphanSample.objects.filter(biosample_id=accession)
122			self.assertTrue(qs.exists())
123
124			async def test_request_with_issues(self) -> None:
125			"""Test a temporary issue with BioSamples reply"""
126
127			with aioresponses() as mocked:
128			mocked.get(
129			'{url}?filter=attr:project:IMAGE&size={size}'.format(
130			url=BIOSAMPLE_URL, size=PAGE_SIZE),
131			status=200,
132			body=page0)
133			mocked.get(
134			'{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
135			url=BIOSAMPLE_URL, size=PAGE_SIZE),
136			status=200,
137			body=issue_page1)
138
139			await check_samples()
140
141			# no objects where tracked since issue in response
142			self.assertEqual(OrphanSample.objects.count(), 0)
143
144			async def test_request_with_html(self) -> None:
145			"""Test a not JSON response (HTML)"""
146
147			with aioresponses() as mocked:
148			mocked.get(
149			'{url}?filter=attr:project:IMAGE&size={size}'.format(
150			url=BIOSAMPLE_URL, size=PAGE_SIZE),
151			status=200,
152			body=page0)
153			mocked.get(
154			'{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
155			url=BIOSAMPLE_URL, size=PAGE_SIZE),
156			status=200,
157			headers={'Content-type': 'text/html'},
158			body="<html>Not a JSON</html>")
159
160			await check_samples()
161
162			# no objects where tracked since issue in response
163			self.assertEqual(OrphanSample.objects.count(), 0)
164
165			async def test_biosamples_down(self) -> None:
166			"""Test a not JSON response (HTML): BioSamples down"""
167
168			with aioresponses() as mocked:
169			mocked.get(
170			'{url}?filter=attr:project:IMAGE&size={size}'.format(
171			url=BIOSAMPLE_URL, size=PAGE_SIZE),
172			status=200,
173			headers={'Content-type': 'text/html'},
174			body="<html>Not a JSON</html>")
175			mocked.get(
176			'{url}?filter=attr:project:IMAGE&page=1&size={size}'.format(
177			url=BIOSAMPLE_URL, size=PAGE_SIZE),
178			status=200,
179			headers={'Content-type': 'text/html'},
180			body="<html>Not a JSON</html>")
181
182			with self.assertRaises(ConnectionError):
183			await check_samples()
184
185			# no objects where tracked since issue in response
186			self.assertEqual(OrphanSample.objects.count(), 0)
187
188			async def test_server_lost(self) -> None:
189			"""Test server disconnect error"""
190
191			with aioresponses() as mocked:
192			mocked.get(
193			'{url}?filter=attr:project:IMAGE&size={size}'.format(
194			url=BIOSAMPLE_URL, size=PAGE_SIZE),
195			exception=ServerDisconnectedError()
196			)
197
198			with self.assertRaises(ConnectionError):
199			await check_samples()
200
201			# no objects where tracked since issue in response
202			self.assertEqual(OrphanSample.objects.count(), 0)
203
204
205			class PurgeOrphanSampleTestCase(BioSamplesMixin, TestCase):
206			fixtures = [
207			'biosample/managedteam',
208			'biosample/orphansample',
209			'uid/dictspecie',
210			]
211
212			def test_purge_orphan_samples(self):
213			"""Test biosample data conversion"""
214
215			with open(os.path.join(DATA_PATH, "SAMEA6376982.json")) as handle:
216			data = json.load(handle)
217
218			self.mock_get.return_value = Mock()
219			self.mock_get.return_value.json.return_value = data
220			self.mock_get.return_value.status_code = 200
221
222			# call my method
223			samples = get_orphan_samples()
224
225			# teams is now a generator
226			self.assertIsInstance(samples, types.GeneratorType)
227			samples = list(samples)
228
229			self.assertEqual(len(samples), 2)
230
231			sample = samples[0]
232			self.assertIsInstance(sample, dict)
233
234			sample = samples[1]
235			self.assertIsInstance(sample, dict)
236
237			# read the team from data
238			team = sample['team']
239			self.assertIsInstance(team, ManagedTeam)
240
241			def test_purge_orphan_samples_not_ready(self):
242			"""Test not ready orphan samples"""
243
244			# Simulate a different status
245			OrphanSample.objects.update(status=SUBMITTED)
246			orphan_count = sum(1 for orphan in get_orphan_samples())
247
248			self.assertEqual(orphan_count, 0)
249
250			def test_purge_orphan_samples_ignore(self):
251			"""Test ignored orphan samples"""
252
253			# Ignoring samples gives no object
254			OrphanSample.objects.update(ignore=True)
255			orphan_count = sum(1 for orphan in get_orphan_samples())
256
257			self.assertEqual(orphan_count, 0)
258
259			def test_purge_orphan_samples_removed(self):
260			"""Test removed orphan samples"""
261
262			# Ignoring samples gives no object
263			OrphanSample.objects.update(removed=True)
264			orphan_count = sum(1 for orphan in get_orphan_samples())
265
266			self.assertEqual(orphan_count, 0)
267
268			def test_purge_orphan_samples_with_limit(self):
269			"""Test get orphan samples with limits"""
270
271			orphan_count = sum(1 for orphan in get_orphan_samples(limit=1))
272			self.assertEqual(orphan_count, 1)
273
274			def test_purge_orphan_private(self):
275			"""Test no access to a BioSamples id (already removed?)"""
276
277			data = {
278			'timestamp': int(time.time() * 1000),
279			'status': 403,
280			'error': 'Forbidden',
281			'exception': (
282			'uk.ac.ebi.biosamples.service.'
283			'BioSamplesAapService$SampleNotAccessibleException'),
284			'message': (
285			'This sample is private and not available for browsing. '
286			'If you think this is an error and/or you should have access '
287			'please contact the BioSamples Helpdesk at biosamples@'
288			'ebi.ac.uk'),
289			'path': '/biosamples/samples/SAMEA6376982'
290			}
291
292			# override mock object
293			self.mock_get.return_value = Mock()
294			self.mock_get.return_value.json.return_value = data
295			self.mock_get.return_value.status_code = 403
296
297			orphan_count = sum(1 for orphan in get_orphan_samples(limit=1))
298			self.assertEqual(orphan_count, 0)
299

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#97)

AsyncBioSamplesTestCase.test_request_with_html() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like