biosample.tests.test_async_biosample.AsyncBioSamplesTestCase.test_request() - Code Metrics - Inspection of ":sparkles: search and track orphaned BioSamples ID..." - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — devel (#90)

by Paolo

created 2020-06-01 20:06 UTC

AsyncBioSamplesTestCase.test_request() A

↳ Parent: biosample.tests.test_async_biosample

Complexity

Conditions

Size

Total Lines	24
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	24
rs	9.5
c	0
b	0
f	0
cc	3
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 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)


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

cnr-ibba / IMAGE-InjectTool

Pull Request — devel (#90)

AsyncBioSamplesTestCase.test_request() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like