|
1 | 1 | import datetime as dt |
2 | 2 | import pickle |
3 | 3 | import socket |
| 4 | +import subprocess |
4 | 5 | from contextlib import suppress |
| 6 | +from pathlib import Path |
5 | 7 |
|
6 | 8 | import pytest |
7 | 9 |
|
@@ -618,3 +620,76 @@ def test_dtls(self): |
618 | 620 | amt = server.write(secret) |
619 | 621 | do_io(src=server, dst=client) |
620 | 622 | assert client.read(amt) == secret |
| 623 | + |
| 624 | + |
| 625 | +@pytest.mark.local |
| 626 | +class TestPrograms: |
| 627 | + @pytest.fixture |
| 628 | + def rootpath(self): |
| 629 | + return Path(__file__).parent.parent |
| 630 | + |
| 631 | + @pytest.fixture |
| 632 | + def tls_server(self, rootpath): |
| 633 | + proc = subprocess.Popen( |
| 634 | + [ |
| 635 | + rootpath / "programs" / "server.py", |
| 636 | + "--tls", |
| 637 | + "--psk-store", |
| 638 | + "cli=secret", |
| 639 | + ] |
| 640 | + ) |
| 641 | + yield proc |
| 642 | + proc.kill() |
| 643 | + proc.wait(1.0) |
| 644 | + |
| 645 | + @pytest.fixture |
| 646 | + def dtls_server(self, rootpath): |
| 647 | + proc = subprocess.Popen( |
| 648 | + [ |
| 649 | + rootpath / "programs" / "server.py", |
| 650 | + "--dtls", |
| 651 | + "--psk-store", |
| 652 | + "cli=secret", |
| 653 | + ] |
| 654 | + ) |
| 655 | + yield proc |
| 656 | + proc.kill() |
| 657 | + proc.wait(1.0) |
| 658 | + |
| 659 | + @pytest.mark.usefixtures("tls_server") |
| 660 | + @pytest.mark.timeout(10) |
| 661 | + def test_e2e_tls(self, rootpath): |
| 662 | + secret = b"a very secret message" |
| 663 | + |
| 664 | + for _ in range(3): |
| 665 | + with subprocess.Popen( |
| 666 | + [ |
| 667 | + rootpath / "programs" / "client.py", |
| 668 | + "--tls", |
| 669 | + "--psk", |
| 670 | + "cli=secret", |
| 671 | + secret, |
| 672 | + ], |
| 673 | + stdout=subprocess.PIPE, |
| 674 | + ) as client: |
| 675 | + out, err = client.communicate() |
| 676 | + assert out == secret + b"\n" |
| 677 | + |
| 678 | + @pytest.mark.usefixtures("dtls_server") |
| 679 | + @pytest.mark.timeout(10) |
| 680 | + def test_e2e_dtls(self, rootpath): |
| 681 | + secret = b"a very secret message" |
| 682 | + |
| 683 | + for _ in range(3): |
| 684 | + with subprocess.Popen( |
| 685 | + [ |
| 686 | + rootpath / "programs" / "client.py", |
| 687 | + "--dtls", |
| 688 | + "--psk", |
| 689 | + "cli=secret", |
| 690 | + secret, |
| 691 | + ], |
| 692 | + stdout=subprocess.PIPE, |
| 693 | + ) as client: |
| 694 | + out, err = client.communicate() |
| 695 | + assert out == secret + b"\n" |
0 commit comments