ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Hard delete applications, nest installer packages, detect PE versions

Browse Source 
Soft delete left orphaned executables on disk and let a deleted name collide
with a fresh import, so delete is now a hard delete that removes the
associated files. The destructive paths (delete, revert-to-record) require a
typed OK because neither can be undone.

Installers keep their filename at import and nest per application/version,
which stops repeat imports doubling up. Version now comes from the same PE
detection the endpoints agent uses; when no version can be read the caller is
asked for one rather than silently storing a blank.

Applications also surface Version and a monitorable toggle, with launch path
and process name required only while monitorable is ticked. Fleet shows the
PC alias, matching PC Details.
This commit is contained in:
benjamin.baker
2026-09-07 15:04:21 -04:00
parent 79545db96e
commit 0134eaa219
17 changed files with 1600 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
tests/test_core/test_application_delete.py
View File

@@ -1,98 +0,0 @@
"""Tests for deactivating an application and what happens to its policy.
Delete here is a soft delete: the row is deactivated, not dropped. That is easy
to forget, because the generated install policy is a separate row that does NOT
follow automatically. Left alone it would keep installing an application the
catalogue reports as gone, and nothing on screen would explain why.
So the deactivation withdraws the policy as well, on the same ownership terms as
unticking the box: ours to remove, theirs to keep.
"""
from shopdb.core.models import Application
from plugins.computers.models import (ConfigurationPolicy, POLICY_SOURCE_LOCAL,
POLICY_SOURCE_AUTOAPP)
from plugins.computers.services.apppolicy import auto_policy_name
def _required_app(client, auth_headers, name='Fleet App'):
response = client.post('/api/applications',
json={'appname': name, 'isrequired': True,
'acknowledgepolicy': True},
headers=auth_headers)
assert response.status_code == 201, response.get_json()
return Application.query.filter_by(appname=name).first()
def test_deleting_an_application_deactivates_rather_than_drops_it(
client, db, auth_headers):
app = Application(appname='Plain App')
db.session.add(app)
db.session.commit()
appid = app.appid
response = client.delete(f'/api/applications/{appid}', headers=auth_headers)
assert response.status_code == 200, response.get_json()
assert db.session.get(Application, appid).isactive is False
def test_deleting_a_required_application_withdraws_its_policy(
client, db, auth_headers):
app = _required_app(client, auth_headers)
assert ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first() is not None
response = client.delete(f'/api/applications/{app.appid}',
headers=auth_headers)
assert response.status_code == 200, response.get_json()
# The policy must not outlive the application it installs.
assert ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first() is None
def test_deleting_an_application_keeps_a_customised_policy(
client, db, auth_headers):
# Same rule as unticking the box: once an admin has edited the policy it is
# theirs, and a delete on another screen must not discard their work.
app = _required_app(client, auth_headers)
policy = ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first()
policy.source = POLICY_SOURCE_LOCAL
db.session.commit()
response = client.delete(f'/api/applications/{app.appid}',
headers=auth_headers)
assert response.status_code == 200, response.get_json()
survivor = ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first()
assert survivor is not None
assert survivor.source == POLICY_SOURCE_LOCAL
def test_deleting_clears_the_required_flag(client, db, auth_headers):
# Otherwise reactivating the application later would show it as required
# while no policy enforces it.
app = _required_app(client, auth_headers)
client.delete(f'/api/applications/{app.appid}', headers=auth_headers)
assert db.session.get(Application, app.appid).isrequired is False
def test_delete_requires_the_delete_permission(client, db, auth_headers,
member_headers):
# Authenticated but unprivileged. The button is hidden for this user in the
# UI; this pins that the endpoint refuses them regardless.
app = Application(appname='Plain App')
db.session.add(app)
db.session.commit()
response = client.delete(f'/api/applications/{app.appid}',
headers=member_headers)
assert response.status_code in (401, 403), response.get_json()
assert db.session.get(Application, app.appid).isactive is not False

139
tests/test_core/test_application_hard_delete.py Normal file
View File

@@ -0,0 +1,139 @@
"""Tests for permanently deleting an application.
This used to be a soft delete, and that was the bug: setting isactive=False
left the row in the table, so the name stayed reserved forever. Creating an
application with the same name failed with "already exists" pointing at a row
no screen would show. Nothing in the UI could explain or clear it.
So delete is now a real delete, and these tests pin the parts that are easy to
get wrong: the dependent rows that would otherwise orphan or block the delete,
the uploaded files that would otherwise linger on disk, and the typed
confirmation that stops it happening on a stray click.
"""
import os
from shopdb.core.models import Application, AppVersion
from plugins.computers.models import ConfigurationPolicy
from plugins.computers.services.apppolicy import auto_policy_name
def _app(db, name='Test App', **kwargs):
app = Application(appname=name, **kwargs)
db.session.add(app)
db.session.commit()
return app
def _delete(client, headers, appid, confirm='OK'):
return client.delete(f'/api/applications/{appid}',
json={'confirm': confirm} if confirm is not None else {},
headers=headers)
def test_delete_removes_the_row_entirely(client, db, auth_headers):
app = _app(db)
appid = app.appid
response = _delete(client, auth_headers, appid)
assert response.status_code == 200, response.get_json()
assert db.session.get(Application, appid) is None
def test_the_name_is_reusable_after_a_delete(client, db, auth_headers):
# The bug this change exists to fix. A soft-deleted row kept its name
# reserved and the conflict named an application the user could not see.
app = _app(db, name='DeviceLogger')
_delete(client, auth_headers, app.appid)
response = client.post('/api/applications',
json={'appname': 'DeviceLogger'},
headers=auth_headers)
assert response.status_code == 201, response.get_json()
def test_delete_is_refused_without_the_typed_confirmation(client, db,
auth_headers):
app = _app(db)
response = _delete(client, auth_headers, app.appid, confirm=None)
assert response.status_code == 400, response.get_json()
assert db.session.get(Application, app.appid) is not None
def test_delete_is_refused_when_the_confirmation_is_wrong_case(client, db,
auth_headers):
# Accepting "ok" would defeat the point of asking for deliberate input.
app = _app(db)
response = _delete(client, auth_headers, app.appid, confirm='ok')
assert response.status_code == 400, response.get_json()
assert db.session.get(Application, app.appid) is not None
def test_delete_takes_the_versions_with_it(client, db, auth_headers):
# appversions.appid is NOT NULL, so leaving these behind would either
# orphan them or block the delete outright.
app = _app(db)
db.session.add(AppVersion(appid=app.appid, version='1.0.0.0'))
db.session.commit()
appid = app.appid
response = _delete(client, auth_headers, appid)
assert response.status_code == 200, response.get_json()
assert AppVersion.query.filter_by(appid=appid).count() == 0
def test_delete_withdraws_the_generated_policy(client, db, auth_headers):
response = client.post('/api/applications',
json={'appname': 'Fleet App', 'isrequired': True,
'acknowledgepolicy': True},
headers=auth_headers)
assert response.status_code == 201, response.get_json()
app = Application.query.filter_by(appname='Fleet App').first()
assert ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first() is not None
_delete(client, auth_headers, app.appid)
assert ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first() is None
def test_delete_removes_the_uploaded_installer(client, db, auth_headers):
app = _app(db)
upload = client.post(f'/api/applications/{app.appid}/package',
data={'file': (__import__('io').BytesIO(b'MZfake'),
'setup.exe')},
content_type='multipart/form-data',
headers=auth_headers)
assert upload.status_code == 200, upload.get_json()
stored = db.session.get(Application, app.appid).installpath
assert stored
from shopdb.core.api.applications import _apppackage_dir
packagedir = os.path.join(_apppackage_dir(), str(app.appid))
# Guard the guard: if the upload did not land where we think, the
# post-delete assertion below would pass without proving anything.
assert os.listdir(packagedir) == ['setup.exe']
_delete(client, auth_headers, app.appid)
# Nothing may remain on disk for this application.
assert not os.path.isdir(packagedir)
def test_delete_requires_the_delete_permission(client, db, member_headers):
app = _app(db)
response = _delete(client, member_headers, app.appid)
assert response.status_code in (401, 403), response.get_json()
assert db.session.get(Application, app.appid) is not None

142
tests/test_core/test_application_monitorable.py Normal file
View File

@@ -0,0 +1,142 @@
"""Tests for the monitorable flag and the two fields it depends on.
A monitorable application tells the agent to watch for a process and relaunch it
when it goes missing. That needs an image name to watch for and a path to run,
so the flag is meaningless without both - a monitor with no process name has
nothing to check, and one with no launch path has nothing to do about it.
The rule lives in _validate_monitorable rather than in a DB constraint so the
columns stay nullable for rows that predate the feature. That makes it a rule
code has to keep choosing to apply, which is what these tests pin: it must hold
on create AND on update, including the update that only ticks the box and sends
nothing else.
"""
from shopdb.core.models import Application
def _app(db, name='Test App', **kwargs):
app = Application(appname=name, **kwargs)
db.session.add(app)
db.session.commit()
return app
def _create(client, headers, **fields):
payload = {'appname': 'Monitored App'}
payload.update(fields)
return client.post('/api/applications', json=payload, headers=headers)
def _update(client, headers, appid, **fields):
return client.put(f'/api/applications/{appid}', json=fields,
headers=headers)
def _message(response):
"""The error text, which sits at data.error.message rather than top level."""
return response.get_json()['data']['error']['message'].lower()
def test_creating_a_monitorable_app_needs_a_process_name(client, auth_headers):
response = _create(client, auth_headers, ismonitorable=True,
monitorlaunchpath=r'C:\Vendor\app.exe')
assert response.status_code == 400
assert 'process name' in _message(response)
def test_creating_a_monitorable_app_needs_a_launch_path(client, auth_headers):
response = _create(client, auth_headers, ismonitorable=True,
monitorprocessname='app.exe')
assert response.status_code == 400
assert 'launch path' in _message(response)
def test_a_monitorable_app_with_both_fields_saves(client, auth_headers, db):
response = _create(client, auth_headers, ismonitorable=True,
monitorprocessname='app.exe',
monitorlaunchpath=r'C:\Vendor\app.exe')
assert response.status_code == 201, response.get_json()
app = Application.query.filter_by(appname='Monitored App').first()
assert app.ismonitorable is True
assert app.monitorprocessname == 'app.exe'
def test_a_non_monitorable_app_needs_neither_field(client, auth_headers):
# The rule must not leak onto ordinary applications, which are the majority.
response = _create(client, auth_headers, ismonitorable=False)
assert response.status_code == 201, response.get_json()
def test_ticking_the_box_alone_is_refused(client, auth_headers, db):
"""The case a payload-only check would miss.
The form sends just the changed field, so an update that ticks monitorable
on a record with no launch path or process name arrives carrying nothing
else. Validating the payload in isolation would see no conflict and let it
through, storing a monitor that cannot run.
"""
app = _app(db, name='Plain App')
response = _update(client, auth_headers, app.appid, ismonitorable=True)
assert response.status_code == 400
# Launch path is checked first, so that is the field named. What matters is
# that the merged state was rejected at all, not which half it complained
# about.
assert 'launch path' in _message(response)
def test_ticking_the_box_is_allowed_when_the_row_already_has_the_fields(
client, auth_headers, db):
# The mirror of the above: validation reads the MERGED result, so stored
# values satisfy it and the operator is not asked to retype them.
app = _app(db, name='Ready App', monitorprocessname='app.exe',
monitorlaunchpath=r'C:\Vendor\app.exe')
response = _update(client, auth_headers, app.appid, ismonitorable=True)
assert response.status_code == 200, response.get_json()
assert Application.query.get(app.appid).ismonitorable is True
def test_clearing_a_field_on_a_monitorable_app_is_refused(client, auth_headers,
db):
# Blanking the process name is the same broken state as never setting it,
# so it has to be refused from this direction too.
app = _app(db, name='Live Monitor', ismonitorable=True,
monitorprocessname='app.exe',
monitorlaunchpath=r'C:\Vendor\app.exe')
response = _update(client, auth_headers, app.appid,
monitorprocessname='')
assert response.status_code == 400
assert 'process name' in _message(response)
def test_whitespace_does_not_count_as_a_process_name(client, auth_headers):
response = _create(client, auth_headers, ismonitorable=True,
monitorprocessname=' ',
monitorlaunchpath=r'C:\Vendor\app.exe')
assert response.status_code == 400
assert 'process name' in _message(response)
def test_unticking_the_box_releases_the_requirement(client, auth_headers, db):
# Turning monitoring off must not be blocked by the fields it needed while
# it was on, or an operator cannot undo a mistake without a round trip.
app = _app(db, name='Live Monitor', ismonitorable=True,
monitorprocessname='app.exe',
monitorlaunchpath=r'C:\Vendor\app.exe')
response = _update(client, auth_headers, app.appid, ismonitorable=False,
monitorprocessname='', monitorlaunchpath='')
assert response.status_code == 200, response.get_json()
assert Application.query.get(app.appid).ismonitorable is False

196
tests/test_core/test_application_revert.py Normal file
View File

@@ -0,0 +1,196 @@
"""Tests for reverting an application back to a plain catalogue record.
Revert is the middle ground between editing and deleting: it withdraws the
generated policy, drops the version rows and removes the uploaded executables,
while the application itself stays in the catalogue.
The line these tests defend is what revert must NOT touch. The warning offers
to remove "policy associations and uploaded executables" and nothing else, so
the image, the name, the description and the knowledgebase links all have to
survive. An operator who reverts to clean up an installer should not discover
afterwards that the record lost its icon.
"""
import io
import os
from shopdb.core.models import Application, AppVersion
from plugins.computers.models import ConfigurationPolicy
from plugins.computers.services.apppolicy import auto_policy_name
def _app(db, name='Test App', **kwargs):
app = Application(appname=name, **kwargs)
db.session.add(app)
db.session.commit()
return app
def _revert(client, headers, appid, confirm='OK'):
return client.post(f'/api/applications/{appid}/revert',
json={'confirm': confirm} if confirm is not None else {},
headers=headers)
def _upload_package(client, headers, appid, name='setup.exe', data=b'MZfake'):
return client.post(f'/api/applications/{appid}/package',
data={'file': (io.BytesIO(data), name)},
content_type='multipart/form-data',
headers=headers)
def _upload_image(client, headers, appid):
# A one-pixel PNG; the route only needs a plausible image file.
png = (b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00'
b'\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc'
b'\x00\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`'
b'\x82')
return client.post(f'/api/applications/{appid}/image',
data={'file': (io.BytesIO(png), 'icon.png')},
content_type='multipart/form-data',
headers=headers)
def test_revert_keeps_the_application_row(client, db, auth_headers):
# The whole point of revert over delete: the record survives.
app = _app(db, name='DeviceLogger')
appid = app.appid
response = _revert(client, auth_headers, appid)
assert response.status_code == 200, response.get_json()
survivor = db.session.get(Application, appid)
assert survivor is not None
assert survivor.appname == 'DeviceLogger'
def test_revert_keeps_the_image(client, db, auth_headers):
# The warning offers to remove policy associations and executables. It says
# nothing about the icon, and a record kept on purpose should keep it.
app = _app(db)
appid = app.appid
assert _upload_image(client, auth_headers, appid).status_code == 200
stored = db.session.get(Application, appid).image
assert stored
response = _revert(client, auth_headers, appid)
assert response.status_code == 200, response.get_json()
assert db.session.get(Application, appid).image == stored
from shopdb.core.api.applications import _appimage_dir
surviving = [name for name in os.listdir(_appimage_dir())
if name.startswith(f'application-{appid}.')]
assert surviving, 'the image file was deleted from disk'
def test_revert_removes_the_uploaded_executables(client, db, auth_headers):
app = _app(db)
appid = app.appid
assert _upload_package(client, auth_headers, appid).status_code == 200
from shopdb.core.api.applications import _apppackage_dir
packagedir = os.path.join(_apppackage_dir(), str(appid))
assert os.listdir(packagedir) == ['setup.exe']
response = _revert(client, auth_headers, appid)
assert response.status_code == 200, response.get_json()
assert not os.path.isdir(packagedir)
reverted = db.session.get(Application, appid)
assert reverted.installpath is None
assert reverted.installersha256 is None
def test_revert_leaves_an_operator_typed_install_path_alone(client, db,
auth_headers):
# A UNC path points at the site's own share, not our upload directory, and
# is not ours to wipe.
share = r'\\fileserver\software\DeviceLogger\setup.exe'
app = _app(db, installpath=share)
response = _revert(client, auth_headers, app.appid)
assert response.status_code == 200, response.get_json()
assert db.session.get(Application, app.appid).installpath == share
def test_revert_takes_the_versions_with_it(client, db, auth_headers):
app = _app(db)
appid = app.appid
db.session.add(AppVersion(appid=appid, version='1.0.0.0'))
db.session.commit()
response = _revert(client, auth_headers, appid)
assert response.status_code == 200, response.get_json()
assert AppVersion.query.filter_by(appid=appid).count() == 0
def test_revert_withdraws_the_generated_policy(client, db, auth_headers):
created = client.post('/api/applications',
json={'appname': 'Fleet App', 'isrequired': True,
'acknowledgepolicy': True},
headers=auth_headers)
assert created.status_code == 201, created.get_json()
app = Application.query.filter_by(appname='Fleet App').first()
assert ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first() is not None
response = _revert(client, auth_headers, app.appid)
assert response.status_code == 200, response.get_json()
assert ConfigurationPolicy.query.filter_by(
name=auto_policy_name(app)).first() is None
assert db.session.get(Application, app.appid).isrequired is False
def test_revert_keeps_the_inventory_links(client, db, auth_headers):
# A machine still has the software installed whether or not we ship an
# installer for it. Dropping these would make the fleet look clean.
from plugins.computers.models import Computer, ComputerInstalledApp
app = _app(db)
computer = Computer(assetid='PC-001')
db.session.add(computer)
db.session.commit()
db.session.add(ComputerInstalledApp(computerid=computer.computerid,
appid=app.appid))
db.session.commit()
response = _revert(client, auth_headers, app.appid)
assert response.status_code == 200, response.get_json()
assert ComputerInstalledApp.query.filter_by(appid=app.appid).count() == 1
def test_revert_is_refused_without_the_typed_confirmation(client, db,
auth_headers):
app = _app(db)
assert _upload_package(client, auth_headers, app.appid).status_code == 200
response = _revert(client, auth_headers, app.appid, confirm=None)
assert response.status_code == 400, response.get_json()
# Nothing may have been removed on a refused call.
from shopdb.core.api.applications import _apppackage_dir
assert os.path.isdir(os.path.join(_apppackage_dir(), str(app.appid)))
def test_revert_is_refused_when_the_confirmation_is_wrong_case(client, db,
auth_headers):
app = _app(db)
response = _revert(client, auth_headers, app.appid, confirm='ok')
assert response.status_code == 400, response.get_json()
def test_revert_requires_the_edit_permission(client, db, member_headers):
app = _app(db)
response = _revert(client, member_headers, app.appid)
assert response.status_code in (401, 403), response.get_json()

12
tests/test_core/test_application_uploads.py
View File

@@ -77,11 +77,15 @@ def test_installer_upload_sets_the_install_path(client, db, auth_headers):
assert response.status_code == 200, response.get_json()
body = response.get_json()['data']
# The installer keeps the name the vendor shipped, namespaced by
# application id so a generic "setup.msi" from two vendors cannot collide.
# The old application-<id>.msi told the operator nothing and made a
# re-import look like new software.
assert body['installpath'] == \
f'/api/applications/package/application-{app.appid}.msi'
# The name the vendor shipped is reported back, since the stored copy is
# renamed and the uploader needs to recognise what they sent.
f'/api/applications/package/{app.appid}/setup.msi'
assert body['uploadedfilename'] == 'setup.msi'
# An MSI is not a PE file, so there is no version resource to read.
assert body['detectedversion'] is None
def test_installer_download_needs_a_token(client, db, auth_headers):
@@ -90,7 +94,7 @@ def test_installer_download_needs_a_token(client, db, auth_headers):
client.post(f'/api/applications/{app.appid}/package',
data={'file': (io.BytesIO(b'fake installer'), 'setup.msi')},
content_type='multipart/form-data', headers=auth_headers)
path = f'/api/applications/package/application-{app.appid}.msi'
path = f'/api/applications/package/{app.appid}/setup.msi'
assert client.get(path).status_code == 401

96
tests/test_core/test_application_version_detection.py Normal file
View File

@@ -0,0 +1,96 @@
"""Tests for reading an installer's version at upload time.
Duplicate application and version rows were being created because an operator
imported the same installer twice: nothing on screen told them the file already
matched a version in the catalogue. Reading the version out of the installer at
upload closes that, and saying so when there is no version to read closes the
rest of it.
Building a real PE with a RT_VERSION resource inline is impractical (see the
agentstore unit tests, which cover the parser itself), so these tests
monkeypatch read_pe_file_version and focus on the wiring: that the upload route
calls it, reports what it found, and reports None rather than guessing.
"""
import io
from shopdb.core.models import Application
from plugins.endpoints.api import agentstore
_PE_BYTES = b'MZ' + b'\x00' * 128
def _app(db, name='Test App'):
app = Application(appname=name)
db.session.add(app)
db.session.commit()
return app
def _upload(client, headers, appid, name='setup.exe', data=_PE_BYTES):
return client.post(f'/api/applications/{appid}/package',
data={'file': (io.BytesIO(data), name)},
content_type='multipart/form-data',
headers=headers)
def test_upload_reports_the_detected_version(client, db, auth_headers,
monkeypatch):
monkeypatch.setattr(agentstore, 'read_pe_file_version',
lambda data: '2.4.1.0')
app = _app(db)
response = _upload(client, auth_headers, app.appid)
assert response.status_code == 200, response.get_json()
assert response.get_json()['data']['detectedversion'] == '2.4.1.0'
def test_upload_reports_none_when_there_is_no_version_resource(
client, db, auth_headers, monkeypatch):
# None is the signal the form uses to ask the operator for a version. It
# must not be substituted with a placeholder, or the duplicate-import
# problem comes straight back.
monkeypatch.setattr(agentstore, 'read_pe_file_version', lambda data: None)
app = _app(db)
response = _upload(client, auth_headers, app.appid)
assert response.status_code == 200, response.get_json()
assert response.get_json()['data']['detectedversion'] is None
def test_detection_does_not_consume_the_stream(client, db, auth_headers,
monkeypatch):
# Version detection reads the upload before it is saved. If it did not
# rewind, the stored installer would be truncated or empty - which would
# not fail any assertion about the version, only corrupt the payload.
monkeypatch.setattr(agentstore, 'read_pe_file_version',
lambda data: '1.0.0.0')
app = _app(db)
assert _upload(client, auth_headers, app.appid).status_code == 200
download = client.get(
f'/api/applications/package/{app.appid}/setup.exe',
headers=auth_headers)
assert download.status_code == 200
assert download.data == _PE_BYTES
def test_detection_failure_does_not_fail_the_upload(client, db, auth_headers,
monkeypatch):
# A version is a convenience; losing it must not cost the operator the
# upload of a 500MB installer.
def _boom(data):
raise ValueError('malformed resource directory')
monkeypatch.setattr(agentstore, 'read_pe_file_version', _boom)
app = _app(db)
response = _upload(client, auth_headers, app.appid)
assert response.status_code == 200, response.get_json()
assert response.get_json()['data']['detectedversion'] is None

156
tests/test_core/test_migration_nest_app_packages.py Normal file
View File

@@ -0,0 +1,156 @@
"""Tests for the migration that nests uploaded installers by application id.
The migration rewrites installpath for every affected row and moves the file on
disk. Both halves matter: a row left on the old scheme is a download that 404s
against a route that no longer serves it, and a file left behind is a download
that 404s against a row pointing at nothing.
These call upgrade()/downgrade() directly with a stubbed bind rather than
running Alembic, so the file movement can be checked against a real temporary
directory without a database round trip.
"""
import importlib.util
import os
import pytest
def _load_migration():
path = os.path.join('migrations', 'versions', '7d46_nest_app_packages.py')
spec = importlib.util.spec_from_file_location('nest_app_packages', path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class _FakeResult:
def __init__(self, rows):
self._rows = rows
def fetchall(self):
return self._rows
class _FakeBind:
"""Records UPDATEs and replays a fixed SELECT, so no DB is needed."""
def __init__(self, rows):
self._rows = rows
self.updates = []
def execute(self, statement, params=None):
text = str(statement)
if text.strip().startswith('SELECT'):
return _FakeResult(self._rows)
self.updates.append(params)
return _FakeResult([])
@pytest.fixture
def migration():
return _load_migration()
def _patch(monkeypatch, migration, bind, directory):
monkeypatch.setattr(migration.op, 'get_bind', lambda: bind)
monkeypatch.setattr(migration, '_package_dir', lambda: directory)
def test_upgrade_moves_the_file_and_repoints_the_row(monkeypatch, migration,
tmp_path):
directory = str(tmp_path)
flat = os.path.join(directory, 'application-7.msi')
with open(flat, 'wb') as handle:
handle.write(b'installer bytes')
bind = _FakeBind([(7, '/api/applications/package/application-7.msi')])
_patch(monkeypatch, migration, bind, directory)
migration.upgrade()
assert not os.path.exists(flat)
moved = os.path.join(directory, '7', 'installer.msi')
assert os.path.isfile(moved)
with open(moved, 'rb') as handle:
assert handle.read() == b'installer bytes'
assert bind.updates == [
{'path': '/api/applications/package/7/installer.msi', 'appid': 7}]
def test_upgrade_repoints_the_row_even_when_the_file_is_missing(
monkeypatch, migration, tmp_path):
# A row left on the old scheme is unserviceable and cannot be repaired by
# re-uploading, so the rewrite must not be conditional on the file.
bind = _FakeBind([(9, '/api/applications/package/application-9.exe')])
_patch(monkeypatch, migration, bind, str(tmp_path))
migration.upgrade()
assert bind.updates == [
{'path': '/api/applications/package/9/installer.exe', 'appid': 9}]
def test_upgrade_leaves_already_nested_rows_alone(monkeypatch, migration,
tmp_path):
# Re-running the migration must not rewrite a path that is already correct,
# which would clobber a real vendor filename with "installer.exe".
bind = _FakeBind([(3, '/api/applications/package/3/DeviceLoggerSetup.exe')])
_patch(monkeypatch, migration, bind, str(tmp_path))
migration.upgrade()
assert bind.updates == []
def test_upgrade_ignores_an_operator_typed_share_path(monkeypatch, migration,
tmp_path):
# The SELECT only matches our own prefix, but prove a UNC path is untouched.
bind = _FakeBind([])
_patch(monkeypatch, migration, bind, str(tmp_path))
migration.upgrade()
assert bind.updates == []
def test_downgrade_puts_the_file_back(monkeypatch, migration, tmp_path):
directory = str(tmp_path)
nested = os.path.join(directory, '7')
os.makedirs(nested)
with open(os.path.join(nested, 'installer.msi'), 'wb') as handle:
handle.write(b'installer bytes')
bind = _FakeBind([(7, '/api/applications/package/7/installer.msi')])
_patch(monkeypatch, migration, bind, directory)
migration.downgrade()
flat = os.path.join(directory, 'application-7.msi')
assert os.path.isfile(flat)
# The emptied directory is cleaned up so a re-upgrade does not find it.
assert not os.path.isdir(nested)
assert bind.updates == [
{'path': '/api/applications/package/application-7.msi', 'appid': 7}]
def test_upgrade_then_downgrade_round_trips(monkeypatch, migration, tmp_path):
directory = str(tmp_path)
with open(os.path.join(directory, 'application-4.exe'), 'wb') as handle:
handle.write(b'payload')
up = _FakeBind([(4, '/api/applications/package/application-4.exe')])
_patch(monkeypatch, migration, up, directory)
migration.upgrade()
down = _FakeBind([(4, up.updates[0]['path'])])
_patch(monkeypatch, migration, down, directory)
migration.downgrade()
restored = os.path.join(directory, 'application-4.exe')
assert os.path.isfile(restored)
with open(restored, 'rb') as handle:
assert handle.read() == b'payload'
assert down.updates[0]['path'] == \
'/api/applications/package/application-4.exe'