import unittest,tempfile,pathlib,subprocess,os,gzip
HOOK=os.environ.get('HOOK',str(pathlib.Path(__file__).with_name('pre-backup.sh')))
class HookTests(unittest.TestCase):
 def setUp(self):
  self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
  self.root=pathlib.Path(self.tmp.name); self.stage=self.root/'stage'; self.stage.mkdir()
  self.old=self.stage/'pg_dumpall.sql.gz'; self.old.write_bytes(b'previous')
  self.bin=self.root/'bin'; self.bin.mkdir()
  self.command('sudo','''shift 2
case "$1" in
 pg_isready) exit 0;;
 pg_dumpall) [ "${FAIL_PG:-0}" = 1 ] && exit 1
 case "$*" in *localhost*) exit 1;; esac
 echo fresh-sql;;
esac
''')
  self.command('mongosh','echo 1')
  self.command('mongodump','mkdir -p "$2"; touch "$2/sample.bson"')
  self.env=dict(os.environ,PATH=str(self.bin)+':'+os.environ['PATH'],RESTIC_STAGE_DIR=str(self.stage))
 def command(self,name,body):
  p=self.bin/name; p.write_text('#!/bin/sh\n'+body+'\n'); p.chmod(0o755)
 def run_hook(self,**env):
  return subprocess.run(['bash',HOOK],env=dict(self.env,**env),capture_output=True,text=True)
 def test_pg_failure_fails_backup_and_preserves_previous_dump(self):
  r=self.run_hook(FAIL_PG='1')
  self.assertNotEqual(r.returncode,0,r.stdout+r.stderr)
  self.assertEqual(self.old.read_bytes(),b'previous')
 def test_successful_socket_dump_replaces_previous(self):
  r=self.run_hook()
  self.assertEqual(r.returncode,0,r.stdout+r.stderr)
  self.assertEqual(gzip.decompress(self.old.read_bytes()),b'fresh-sql\n')
  self.assertTrue((self.stage/'mongodump/sample.bson').exists())
 def test_mongo_failure_fails_backup_and_preserves_previous(self):
  d=self.stage/'mongodump'; d.mkdir(); (d/'old.bson').write_bytes(b'previous')
  self.command('mongodump','exit 1')
  r=self.run_hook()
  self.assertNotEqual(r.returncode,0,r.stdout+r.stderr)
  self.assertEqual((d/'old.bson').read_bytes(),b'previous')
if __name__=='__main__':unittest.main()