Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _update_apps method in pyatom

Best Python code snippet using pyatom_python

test_remoterepository.py

Source:test_remoterepository.py

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()...

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.