"""Pruebas de los simuladores y del checksum: no necesitan ninguna topología.""" import unittest import rdt_sim import seq_space import udp_checksum import window_sim class A_BuildingUp(unittest.TestCase): MESSAGES = rdt_sim.MESSAGES def run_it(self, name): return rdt_sim.run_scenario(name) def test_v1_works_on_a_perfect_channel(self): self.assertEqual(self.run_it("v1-perfecto").delivered, self.MESSAGES) def test_v1_delivers_garbage_without_noticing(self): result = self.run_it("v1-errores") self.assertIn("?!?", result.delivered) self.assertTrue(result.finished) def test_v2_recovers_from_damaged_data(self): self.assertEqual(self.run_it("v2-datos").delivered, self.MESSAGES) def test_v2_duplicates_when_the_ack_is_damaged(self): self.assertEqual(self.run_it("v2-ack").delivered, ["a", "a", "b", "c"]) def test_the_sequence_bit_fixes_it(self): self.assertEqual(self.run_it("v21-ack").delivered, self.MESSAGES) def test_no_nak_is_needed(self): result = self.run_it("v22-datos") self.assertEqual(result.delivered, self.MESSAGES) self.assertFalse(any("NAK" in text for _, _, text in result.trace)) def test_without_a_timer_a_loss_hangs_forever(self): result = self.run_it("v22-perdida") self.assertFalse(result.finished) self.assertEqual(result.delivered, ["a"]) def test_v3_survives_loss_in_both_directions_and_a_late_ack(self): for name in ("v3-dato", "v3-ack", "v3-tarde"): with self.subTest(name): result = self.run_it(name) self.assertTrue(result.finished) self.assertEqual(result.delivered, self.MESSAGES) def test_a_timer_without_sequence_numbers_duplicates(self): self.assertEqual(self.run_it("v3-sin-seq").delivered, ["a", "a", "b", "c"]) class B_Traces(unittest.TestCase): """La pregunta de repaso: ventana 4, se pierde el 2 de los paquetes 0 a 5.""" def trace(self, protocol): trace = [] config = window_sim.Config(protocol=protocol, window=4, packets=6, drop_data=[(2, 1)]) return window_sim.WindowSim(config, trace).run(), trace def test_gbn_resends_everything_from_the_lost_packet(self): stats, trace = self.trace("gbn") self.assertEqual(stats.delivered, list(range(6))) self.assertEqual(stats.transmissions, 10) repeated = [text for _, _, text in trace if text.startswith("repite")] self.assertEqual(repeated, ["repite 2", "repite 3", "repite 4", "repite 5"]) def test_gbn_receiver_discards_and_repeats_the_last_ack(self): _, trace = self.trace("gbn") discarded = [text for _, _, text in trace if "lo descarta" in text] self.assertEqual(len(discarded), 3) def test_sr_resends_only_the_lost_packet(self): stats, trace = self.trace("sr") self.assertEqual(stats.delivered, list(range(6))) self.assertEqual(stats.transmissions, 7) repeated = [text for _, _, text in trace if text.startswith("repite")] self.assertEqual(repeated, ["repite 2"]) def test_sr_delivers_in_order_even_when_it_receives_out_of_order(self): _, trace = self.trace("sr") self.assertTrue(any("entrega el contenido [2, 3, 4, 5]" in text for _, _, text in trace)) class C_Utilization(unittest.TestCase): def test_the_week_example(self): d = 1500 * 8 / 1e6 # ms a 1 Gbps expected = window_sim.stop_and_wait_utilization(d, 30) self.assertAlmostEqual(expected, 0.0004, places=5) stats = window_sim.simulate(protocol="gbn", window=1, packets=100, d=d, rtt=30) self.assertAlmostEqual(stats.utilization(d), expected, places=6) self.assertEqual(window_sim.window_to_fill(d, 30), 2501) def test_the_review_question(self): d = 1500 * 8 / 1e5 # ms a 100 Mbps self.assertAlmostEqual(window_sim.stop_and_wait_utilization(d, 40), 0.00299, places=5) self.assertEqual(window_sim.window_to_fill(d, 40), 335) def test_utilization_grows_with_the_window_until_it_fills_the_link(self): d, rtt = 1.0, 40.0 for window, packets in [(1, 200), (10, 400), (20, 800), (41, 1600), (80, 3200)]: stats = window_sim.simulate(protocol="sr", window=window, packets=packets, d=d, rtt=rtt) self.assertAlmostEqual(stats.utilization(d), window_sim.utilization(window, d, rtt), delta=0.05, msg=f"ventana {window}") class D_Comparison(unittest.TestCase): @classmethod def setUpClass(cls): cls.both = window_sim.sweep(window=30, packets=1000, seeds=3, losses=(0, 0.01, 0.1)) cls.acks = window_sim.sweep(window=30, packets=1000, seeds=3, losses=(0.05,), ack_only=True) def test_without_loss_they_are_the_same(self): row = self.both[0] self.assertEqual(row["gbn"], row["sr"]) self.assertEqual(row["gbn"][0], 1.0) def test_with_loss_gbn_sends_more_and_the_gap_widens(self): low, high = self.both[1], self.both[2] self.assertGreater(low["gbn"][0], low["sr"][0]) self.assertGreater(high["gbn"][0] / high["sr"][0], low["gbn"][0] / low["sr"][0]) def test_loss_of_acks_alone_favors_the_cumulative_ack(self): row = self.acks[0] self.assertLess(row["gbn"][0], row["sr"][0]) self.assertGreater(row["gbn"][1], row["sr"][1]) class E_SequenceSpace(unittest.TestCase): def test_sr_with_four_numbers_and_window_three_delivers_old_data_as_new(self): stats, _ = window_sim.ambiguity_run("sr", 4, 3) self.assertNotEqual(stats.delivered, list(range(len(stats.delivered)))) def test_sr_with_four_numbers_and_window_two_is_correct(self): stats, _ = window_sim.ambiguity_run("sr", 4, 2) self.assertTrue(stats.finished) self.assertEqual(stats.delivered, list(range(8))) def test_gbn_limits_by_simulation(self): good, _ = window_sim.ambiguity_run("gbn", 4, 3) self.assertEqual(good.delivered, list(range(8))) bad, _ = window_sim.ambiguity_run("gbn", 4, 4) self.assertNotEqual(bad.delivered, list(range(len(bad.delivered)))) def test_the_limits_for_every_size(self): for bits in range(1, 8): modulus = 2 ** bits self.assertEqual(seq_space.largest_safe_window("gbn", modulus), modulus - 1) self.assertEqual(seq_space.largest_safe_window("sr", modulus), modulus // 2) class F_Checksum(unittest.TestCase): def test_the_datagram_of_the_notes(self): value = udp_checksum.udp_checksum("10.209.21.10", "10.209.23.10", 5000, 9009, b"hola") self.assertEqual(value, 0xb296) def test_odd_length_is_padded(self): even = udp_checksum.udp_checksum("10.0.0.1", "10.0.0.2", 1, 2, b"ab\0") odd = udp_checksum.udp_checksum("10.0.0.1", "10.0.0.2", 1, 2, b"ab") # El relleno no cuenta en el largo: los dos datos solo difieren en esa palabra de largo. self.assertEqual(sum(udp_checksum.words(b"ab\0")), sum(udp_checksum.words(b"ab"))) self.assertNotEqual(even, odd) def test_a_computed_zero_is_sent_as_all_ones(self): base = udp_checksum.ones_complement_sum(udp_checksum.words( udp_checksum.pseudo_header("10.0.0.1", "10.0.0.2", 10) + bytes([0, 1, 0, 2, 0, 10, 0, 0]))) filler = (0xffff - base) & 0xffff value = udp_checksum.udp_checksum("10.0.0.1", "10.0.0.2", 1, 2, filler.to_bytes(2, "big")) self.assertEqual(value, 0xffff) def test_swapping_two_words_goes_undetected(self): first = udp_checksum.udp_checksum("10.0.0.1", "10.0.0.2", 1, 2, b"abcd") second = udp_checksum.udp_checksum("10.0.0.1", "10.0.0.2", 1, 2, b"cdab") self.assertEqual(first, second) if __name__ == "__main__": unittest.main()