Simulate file upload operations
(self, target_ip, target_port, duration)
| 437 | time.sleep(1.0 / self.rate_limit) |
| 438 | |
| 439 | def file_upload_simulation(self, target_ip, target_port, duration): |
| 440 | """Simulate file upload operations""" |
| 441 | end_time = time.time() + duration |
| 442 | session = self.create_http_session() |
| 443 | scheme = "https" if target_port == 443 else "http" |
| 444 | base_url = f"{scheme}://{target_ip}:{target_port}" |
| 445 | |
| 446 | while time.time() < end_time and self.running: |
| 447 | try: |
| 448 | # Generate random file data |
| 449 | file_size = random.randint(1024, 10240) # 1KB to 10KB |
| 450 | file_data = os.urandom(file_size) |
| 451 | files = {'file': ('test_file.jpg', file_data, 'image/jpeg')} |
| 452 | |
| 453 | start_time = time.time() |
| 454 | response = session.post(f"{base_url}/api/v1/upload", files=files, timeout=self.timeout) |
| 455 | response_time = (time.time() - start_time) * 1000 |
| 456 | |
| 457 | self.stats['requests_sent'] += 1 |
| 458 | self.stats['total_bytes'] += file_size |
| 459 | self.stats['response_times'].append(response_time) |
| 460 | |
| 461 | if 200 <= response.status_code < 400: |
| 462 | self.stats['requests_successful'] += 1 |
| 463 | else: |
| 464 | self.stats['requests_failed'] += 1 |
| 465 | |
| 466 | except Exception as e: |
| 467 | self.stats['requests_failed'] += 1 |
| 468 | self.stats['requests_sent'] += 1 |
| 469 | |
| 470 | time.sleep(2.0 / self.rate_limit) # Slower rate for uploads |
| 471 | |
| 472 | def mixed_workload_test(self, target_ip, target_port, duration): |
| 473 | """Combined workload simulating real user behavior""" |
nothing calls this directly
no test coverage detected