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

Completed

Pull Request — devel (#90)

by Paolo

created 2020-06-01 18:03 UTC

AsyncBioSamplesTestCase.setUp() A

↳ Parent: biosample.tests.test_async_biosample

Complexity

Conditions

Size

Total Lines	18
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	11
dl	0
loc	18
rs	9.85
c	0
b	0
f	0
cc	1
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 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)


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

cnr-ibba / IMAGE-InjectTool

Pull Request — devel (#90)

AsyncBioSamplesTestCase.setUp() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like