mirror of
https://codeberg.org/privacy1st/subprocess-util
synced 2025-04-02 11:16:00 +02:00
129 lines
3.4 KiB
Python
129 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from subprocess_util.exec_capture import execute_capture
|
|
from subprocess_util.exec_print_capture import execute_print_capture
|
|
from subprocess_util.exec_produce_chunks import execute_produce_chunks
|
|
from subprocess_util.chunked_transfer.active_send.exec_receive_chunks_passive import exec_receive_chunks_passive
|
|
from subprocess_util.chunked_transfer.active_send.exec_send_chunks_active import exec_send_chunks_active
|
|
|
|
|
|
def test():
|
|
test16()
|
|
|
|
|
|
def _init(test_number: int):
|
|
print(f"TEST {test_number}")
|
|
test_dir = Path('test')
|
|
if test_dir.exists():
|
|
shutil.rmtree('test')
|
|
Path('test').mkdir(exist_ok=False)
|
|
|
|
|
|
def test1():
|
|
_init(1)
|
|
|
|
returncode, out, err = execute_capture(['ls', '-la'])
|
|
print(f'stdout:\n{out}\nstderr:\n{err}')
|
|
print()
|
|
returncode, out, err = execute_capture(['ls', '/foo/bar'])
|
|
print(f'stdout:\n{out}\nstderr:\n{err}')
|
|
|
|
|
|
def test2():
|
|
_init(2)
|
|
|
|
_returncode, _out, _err = execute_print_capture(['ls', '-la'])
|
|
print()
|
|
_returncode, _out, _err = execute_print_capture(['ls', '/foo/bar'])
|
|
|
|
|
|
def test10():
|
|
_init(10)
|
|
|
|
source_file = Path('src/subprocess_util/exec_capture.py')
|
|
chunk_dir = Path('test')
|
|
target_file = Path('test/exec_capture.py')
|
|
|
|
def get_chunk_path(chunk_no: int):
|
|
return chunk_dir.joinpath(f'{source_file.name}.CHUNK.{chunk_no}')
|
|
|
|
def handle_chunk(chunk_no: int, _last_chunk: bool):
|
|
chunk_path = get_chunk_path(chunk_no)
|
|
print(f'Handling chunk {chunk_path}')
|
|
|
|
# Read chunk.
|
|
chunk = chunk_path.read_bytes()
|
|
# Append chunk to target.
|
|
with target_file.open("ab") as f:
|
|
f.write(chunk)
|
|
# Delete chunk.
|
|
chunk_path.unlink(missing_ok=False)
|
|
|
|
execute_produce_chunks(
|
|
command=['cat', str(source_file)],
|
|
get_chunk_path=get_chunk_path,
|
|
handle_chunk=handle_chunk,
|
|
chunk_size=512,
|
|
)
|
|
|
|
|
|
def test15_recieve():
|
|
"""
|
|
First start the passive side.
|
|
"""
|
|
exec_receive_chunks_passive(
|
|
command=['tee', '/home/yoda/test/15.txt'],
|
|
target_chunk_dir=Path('/home/yoda/test/15'),
|
|
)
|
|
|
|
|
|
def test15_send():
|
|
"""
|
|
Then run the active side.
|
|
"""
|
|
exec_send_chunks_active(
|
|
command=['ls', '/home'],
|
|
ssh_target='nas',
|
|
source_chunk_dir=Path('test/15'),
|
|
target_chunk_dir=Path('/home/yoda/test/15'),
|
|
)
|
|
|
|
|
|
def test16():
|
|
"""
|
|
Minimal example to replicate a BTRFS subvolume with an active sending side.
|
|
"""
|
|
snapshot = 'blogger.privacy1st.de/20230213T1809_u/'
|
|
|
|
child_path = '/mnt/backup/snap/' + snapshot
|
|
ssh_target = 'rootnas'
|
|
target_path = '/mnt/data/test/' + snapshot
|
|
|
|
print(f'nas: '
|
|
f'sudo btrfs-receive-chunks-passive {target_path}')
|
|
print(f'pi3bplus: '
|
|
f'sudo btrfs-send-chunks-active {child_path} {ssh_target} {target_path}')
|
|
|
|
|
|
def test17():
|
|
"""
|
|
Minimal example to replicate a BTRFS subvolume with an active receiving side.
|
|
"""
|
|
snapshot = 'blogger.privacy1st.de/20230213T1809_u/'
|
|
|
|
child_path = '/mnt/data/snap/' + snapshot
|
|
ssh_source = 'rootnas'
|
|
target_path = '/mnt/backup/test/' + snapshot
|
|
|
|
print(f'nas: '
|
|
f'sudo btrfs-send-chunks-passive {child_path}')
|
|
print(f'pi3bplus: '
|
|
f'sudo btrfs-receive-chunks-active --child={child_path} {ssh_source} {target_path}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
test()
|