57 lines
2.2 KiB
Python
57 lines
2.2 KiB
Python
import hashlib
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
s=importlib.util.spec_from_file_location('expand',Path(__file__).with_name('expand_forge_profiles.py'))
|
|
m=importlib.util.module_from_spec(s)
|
|
s.loader.exec_module(m)
|
|
|
|
|
|
class AtomicOwnership(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory()
|
|
self.path=Path(self.tmp.name)/'repo/hooks/pre-receive.d/10-test'
|
|
self.path.parent.mkdir(parents=True)
|
|
self.body=b'#!/bin/sh\nexit 0\n'
|
|
self.digest=hashlib.sha256(self.body).hexdigest()
|
|
self.saved=m.t.native.execute
|
|
|
|
def execute(pod,*args,payload=None,**kwargs):
|
|
result=subprocess.run(args,input=payload,capture_output=True,timeout=5)
|
|
if result.returncode:raise ValueError('refused')
|
|
return result.stdout
|
|
|
|
m.t.native.execute=execute
|
|
|
|
def tearDown(self):
|
|
m.t.native.execute=self.saved
|
|
self.tmp.cleanup()
|
|
|
|
def test_install_and_owned_rollback(self):
|
|
m.install({},str(self.path),self.body,self.digest)
|
|
self.assertEqual(self.path.read_bytes(),self.body)
|
|
self.assertEqual(self.path.stat().st_mode&0o777,0o755)
|
|
m.remove_owned({},str(self.path),self.digest,require_marker=True)
|
|
self.assertFalse(self.path.exists())
|
|
|
|
def test_existing_destination_never_overwritten(self):
|
|
self.path.write_bytes(b'preexisting')
|
|
with self.assertRaises(ValueError):m.install({},str(self.path),self.body,self.digest)
|
|
self.assertEqual(self.path.read_bytes(),b'preexisting')
|
|
|
|
def test_changed_content_not_removed(self):
|
|
m.install({},str(self.path),self.body,self.digest)
|
|
self.path.write_bytes(b'changed')
|
|
with self.assertRaises(ValueError):m.remove_owned({},str(self.path),self.digest,require_marker=True)
|
|
self.assertTrue(self.path.exists())
|
|
|
|
def test_identical_unowned_file_not_removed_during_recovery(self):
|
|
self.path.write_bytes(self.body)
|
|
with self.assertRaises(ValueError):m.remove_owned({},str(self.path),self.digest,require_marker=True)
|
|
self.assertTrue(self.path.exists())
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|