How to use _update_apps method in pyatom

Best Python code snippet using pyatom_python

test_remoterepository.py

Source:test_remoterepository.py Github

copy

Full Screen

1import os2from datetime import datetime, timezone3from unittest.mock import Mock, patch4import background_task5import requests6from background_task.tasks import Task7from django.conf import settings8from django.contrib.auth.models import User9from repomaker.models import App, RemoteApp, Repository, \10 RemoteRepository11from repomaker.models.repository import AbstractRepository12from repomaker.storage import get_remote_repo_path13from repomaker.tasks import PRIORITY_REMOTE_REPO14from requests.exceptions import HTTPError15from .. import RmTestCase16class RemoteRepositoryTestCase(RmTestCase):17 def setUp(self):18 self.repo = RemoteRepository.objects.get(pk=1)19 def test_get_path(self):20 self.assertEqual(os.path.join(settings.MEDIA_ROOT, 'remote_repo_1'), self.repo.get_path())21 with self.assertRaises(NotImplementedError):22 AbstractRepository().get_repo_path()23 def test_initial_update(self):24 """25 Makes sure that pre-installed remote repositories will be updated on first start.26 """27 tasks = Task.objects.all()28 self.assertTrue(len(tasks) >= 1)29 for task in tasks:30 self.assertEqual('repomaker.tasks.update_remote_repo', task.task_name)31 @patch('repomaker.tasks.update_remote_repo')32 def test_update_async(self, update_remote_repo):33 """34 Makes sure that the asynchronous update starts a background task.35 """36 self.repo.update_scheduled = False37 self.repo.update_async()38 update_remote_repo.assert_called_once_with(self.repo.id, repeat=Task.DAILY,39 priority=PRIORITY_REMOTE_REPO)40 self.assertTrue(self.repo.update_scheduled)41 @patch('repomaker.tasks.update_remote_repo')42 def test_update_async_not_called_when_update_scheduled(self, update_remote_repo):43 """44 Makes sure that the asynchronous update does not start another background task45 when another one is scheduled already.46 """47 self.repo.update_scheduled = True48 self.repo.update_async()49 self.assertFalse(update_remote_repo.called)50 @patch('repomaker.models.remoterepository.RemoteRepository._update_apps')51 @patch('fdroidserver.index.download_repo_index')52 def test_update_index_only_when_new(self, download_repo_index, _update_apps):53 """54 Test that a remote repository is only updated when the index changed since last time.55 """56 download_repo_index.return_value = {57 'repo': {'name': 'Test Name', 'timestamp': 0}58 }, 'etag'59 self.repo.last_change_date = datetime.now(tz=timezone.utc)60 self.repo.update_index(update_apps=True)61 download_repo_index.assert_called_once_with(self.repo.get_fingerprint_url(), etag=None)62 self.assertNotEqual('Test Name', self.repo.name)63 self.assertFalse(_update_apps.called)64 @patch('repomaker.models.remoterepository.RemoteRepository._update_apps')65 @patch('fdroidserver.index.download_repo_index')66 def test_update_index_only_when_not_none(self, download_repo_index, _update_apps):67 """68 Test that a remote repository is only updated when the index changed since last time.69 """70 download_repo_index.return_value = None, 'etag'71 self.repo.update_index(update_apps=True)72 download_repo_index.assert_called_once_with(self.repo.get_fingerprint_url(), etag=None)73 self.assertFalse(_update_apps.called)74 @patch('requests.get')75 @patch('repomaker.models.remoterepository.RemoteRepository._update_apps')76 @patch('fdroidserver.index.download_repo_index')77 def test_update_index(self, download_repo_index, _update_apps, get):78 """79 Test that a remote repository is updated and a new icon is downloaded.80 """81 # return a fake index82 download_repo_index.return_value = {83 'repo': {84 'name': 'Test Name',85 'description': 'Test <script>Description',86 'icon': 'test-icon.png',87 'timestamp': datetime.utcnow().timestamp() * 1000,88 'mirrors': [89 'mirror1',90 'mirror2',91 ],92 },93 'apps': [],94 'packages': [],95 }, 'etag'96 # fake return value of GET request for repository icon97 get.return_value.status_code = requests.codes.ok98 get.return_value.content = b'foo'99 # update index and ensure it would have been downloaded100 repo = self.repo101 repo.update_index(update_apps=True)102 download_repo_index.assert_called_once_with(repo.get_fingerprint_url(), etag=None)103 # assert that the repository metadata was updated with the information from the index104 self.assertEqual('Test Name', repo.name)105 self.assertEqual('Test Description', repo.description)106 self.assertEqual('["mirror1", "mirror2"]', repo.mirrors)107 # assert that new repository icon was downloaded and changed108 get.assert_called_once_with(repo.url + '/icons/' + 'test-icon.png',109 headers={'User-Agent': 'F-Droid'})110 self.assertEqual(os.path.join(get_remote_repo_path(repo), 'test-icon.png'), repo.icon.name)111 # assert that an attempt was made to update the apps112 self.assertTrue(_update_apps.called)113 @patch('repomaker.models.remoteapp.RemoteApp.update_from_json')114 @patch('fdroidserver.index.download_repo_index')115 def test_update_index_retry_when_failed(self, download_repo_index, update_from_json):116 """117 Test that a remote repository is only updated when the index changed since last time.118 """119 # return a fake index120 index = {121 'repo': {122 'name': 'Test Name',123 'description': 'Test Description',124 'timestamp': datetime.utcnow().timestamp() * 1000,125 },126 'apps': [127 {128 'packageName': 'org.example',129 'name': 'test app',130 'lastUpdated': datetime.utcnow().timestamp() * 1000,131 },132 ],133 'packages': {134 'org.example': []135 },136 }137 download_repo_index.return_value = index, 'etag'138 update_from_json.side_effect = HTTPError(Mock(status=502))139 with self.assertRaises(HTTPError):140 self.repo.update_index(update_apps=True)141 download_repo_index.assert_called_once_with(self.repo.get_fingerprint_url(), etag=None)142 update_from_json.assert_called_once_with(index['apps'][0])143 self.assertEqual(datetime.fromtimestamp(0, timezone.utc), self.repo.last_change_date)144 self.assertIsNone(self.repo.index_etag)145 @patch('fdroidserver.net.http_get')146 def test_update_icon_without_pk(self, http_get):147 # create unsaved repo without primary key148 repo = RemoteRepository(url='http://test', last_change_date=datetime.now(tz=timezone.utc))149 # update icon150 http_get.return_value = b'icon-data', 'new_etag'151 repo._update_icon('test.png') # pylint: disable=protected-access152 http_get.assert_called_once_with('http://test/icons/test.png', None)153 # assert that there is no `None` in the path, but the repo number (primary key)154 self.assertFalse('None' in repo.icon.name)155 self.assertTrue(str(repo.pk) in repo.icon.name)156 @patch('fdroidserver.index.download_repo_index')157 def test_fail(self, download_repo_index):158 # assert that pre-installed remote repository is initially enabled159 self.assertFalse(self.repo.disabled)160 # get initial update task from pre-installed remote repository161 all_tasks = Task.objects.all()162 self.assertEqual(2, all_tasks.count())163 task = all_tasks[0]164 # set initial update task to be one short of the maximum number of attempts165 task.attempts = settings.MAX_ATTEMPTS - 1166 task.save()167 # run the task and provide it with malformed data, so it fails168 download_repo_index.return_value = 'malformed', None169 background_task.tasks.tasks.run_next_task()170 self.assertEqual(1, download_repo_index.call_count)171 # ensure that remote repository got disabled172 self.repo = RemoteRepository.objects.get(pk=1)173 self.assertTrue(self.repo.disabled)174 def test_remove_old_apps(self):175 RemoteApp.objects.create(repo=self.repo, package_id="delete",176 last_updated_date=self.repo.last_updated_date)177 RemoteApp.objects.create(repo=self.repo, package_id="do not delete",178 last_updated_date=self.repo.last_updated_date)179 packages = ["do not delete"]180 self.repo._remove_old_apps(packages) # pylint: disable=protected-access181 # assert that only the expected app was deleted182 self.assertEqual(1, RemoteApp.objects.count())183 self.assertEqual("do not delete", RemoteApp.objects.all()[0].package_id)184 def test_remove_old_apps_thousands(self):185 """186 Tests that we do not run into sqlite3.OperationalError: too many SQL variables187 """188 RemoteApp.objects.create(repo=self.repo, package_id="delete",189 last_updated_date=self.repo.last_updated_date)190 # add lots of fake packages191 packages = []192 for i in range(1, 3000):193 packages.append(str(i))194 self.repo._remove_old_apps(packages) # pylint: disable=protected-access195 # assert that all remote apps could be deleted196 self.assertFalse(RemoteApp.objects.exists())197 def test_is_in_repo(self):198 repo = Repository.objects.create(user=User.objects.create_user('user2'))199 app = App.objects.create(repo=repo, package_id="org.example")200 remote_app = RemoteApp.objects.create(repo=self.repo, package_id="org.example",201 last_updated_date=self.repo.last_updated_date)202 self.assertTrue(remote_app.is_in_repo(repo))203 app.package_id = "different"204 app.save()...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run pyatom automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful