Learn UC Collaboration

In Part 1 we covered SIPp from the ground up – installation, XML scenarios, RTP audio, output options, and CSV injection for multi-number testing. By the end of that post you could run SIPp standalone from the command line to test any SIP extension.

That’s a solid foundation. But running SIPp manually doesn’t scale. If you want to test 50 extensions on a schedule, get structured pass/fail results, or build phone testing into a larger application or CI pipeline, you need to drive SIPp from code.

In this post we’ll do exactly that. We’ll wrap SIPp in Python, parse its output and build a class you can drop into any project.

Why Python?
Running SIPp from Python
1. Architecture
Using the Test Runner Class
1. Sample Terminal Output
2. Sample JSON Output Data
Scheduling Automated Tests
YouTube Explainer
What’s Next ?

Why Python?

SIPp is a command-line tool. Python’s subprocess module can launch any command-line tool, capture its output, and react to its exit code which makes it a natural fit. Beyond just launching SIPp, Python lets us:

Parse SIPp’s CSV statistics files to get structured pass/fail data
Read the SIP message logs to understand exactly what happened on a failed call
Loop over a list of extensions and aggregate results
Save results to JSON for reporting or further processing
Integrate with scheduling tools, web frameworks, or CI systems

Running SIPp from Python

To integrate SIPp with Python, we are going to write a class called SIPpTestRunner that handles the entire workflow end to end. You instantiate it once with your target CUCM IP, scenario file, and a few other details, and from there it takes care of everything. This includes launching SIPp with the right flags, parsing the stats CSV, reading the SIP message logs to determine the actual call outcome, and saving results to JSON & presenting them on screen. Whether you’re testing a single extension or a batch of hundred extensions, the interface stays exactly the same.

Architecture

Here is the high level architectural overview of how different components of the class interact with each other to programmatically complete the end-to-end execution of the SIPp testing process.

Here is a brief overview of different function components of this class. You can keep this in mind while going through the architecture diagram & correlate it with the actual code given below.

__init__ : Sets up the runner object with your target details – CUCM IP, scenario file path, local IP, and where to write the logs.
run_call : This is the main workhorse. It builds the SIPp command with the right flags, fires it as a subprocess, waits for it to finish, and then hands the output off to the parsing methods. It returns a single dict with everything about that call’s outcome.
run_bulk_test : This function loops over a list of extensions, calls run_call for each one with a configurable delay in between, and returns a list of result dicts. This is what you use when you need to test a batch of numbers in one shot.
_parse_stats : This reads the CSV file SIPp writes and pulls out the key counters from the final cumulative row like successful calls, failed calls, response codes etc.
_parse_messages : Reads the raw SIP message log and extracts the actual response codes received during the call, turning them into meaningful boolean flags like call_answered, busy, not_found, rejected.
_interpret : Takes the flags from _parse_messages and turns them into a single human readable outcome string like ANSWERED, BUSY, REJECTED etc.
save_results : Writes the full results list out to a JSON file. It’s useful for keeping a record of test runs or feeding the data into another system downstream.
print_summary : Prints a formatted table to the terminal showing extension, status, and outcome for every call in the results list along with a pass/fail count at the bottom.

import subprocess
import os
import csv
import re
import json
import time
import logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)
class SIPpTestRunner:
    def __init__(self, scenario_file, target_ip, log_dir, local_ip=None,
                 sipp_path="sipp",):
        self.scenario_file = scenario_file
        self.target_ip = target_ip
        self.local_ip = local_ip
        self.sipp_path = sipp_path
        self.log_dir = log_dir
        os.makedirs(log_dir, exist_ok=True)
    def run_call(self, extension, duration_ms=10000, timeout_s=30):
        """
        Place a single test call to the given extension.
        Returns a dict with the full test result.
        """
        stats_file = os.path.join(self.log_dir, f"{extension}_stats.csv")
        msg_log    = os.path.join(self.log_dir, f"{extension}_messages.log")
        err_log    = os.path.join(self.log_dir, f"{extension}_errors.log")
        cmd = [
            self.sipp_path,
            self.target_ip,
            "-sf", self.scenario_file,
            "-s", extension,
            "-m", "1",
            "-d", str(duration_ms),
            "-timeout", str(timeout_s),
            "-trace_stat", "-stf", stats_file,
            "-trace_msg",  "-message_file", msg_log,
            "-trace_err",  "-error_file", err_log,
            "-nostdin"
        ]
        logger.info(f"Testing extension {extension}...")
        try:
            proc = subprocess.run(
                cmd, capture_output=True, text=True,
                timeout=timeout_s + 15
            )
            return_code = proc.returncode
        except subprocess.TimeoutExpired:
            logger.error(f"Process timed out for {extension}")
            return {
                "extension": extension,
                "status": "TIMEOUT",
                "outcome": "TIMEOUT",
                "return_code": -1
            }
        # Building the result dict
        result = {
            "extension": extension,
            "return_code": return_code,
            "status": "PASS" if return_code == 0 else "FAIL"
        }
        # Adding stats if available
        if os.path.exists(stats_file):
            result.update(self._parse_stats(stats_file))
        # Analyzing Messages if available
        if os.path.exists(msg_log):
            msg_data = self._parse_messages(msg_log)
            result.update(msg_data)
            result["outcome"] = self._interpret(msg_data)
        logger.info(f"{extension}: {result.get('outcome', result['status'])}")
        return result
    def run_bulk_test(self, extensions, delay_between_calls=2):
        """
        Testing a list of extensions sequentially with delay between calls.
        Returns a list of result dicts.
        """
        results = []
        for ext in extensions:
            result = self.run_call(ext)
            results.append(result)
            if delay_between_calls > 0:
                time.sleep(delay_between_calls)
        return results
    def save_results(self, results, output_file="test_results.json"):
        with open(output_file, "w") as f:
            json.dump(results, f, indent=2)
        logger.info(f"Results saved to {output_file}")
    def print_summary(self, results):
        print("\n" + "=" * 60)
        print(f"{'EXTENSION':<15} {'STATUS':<10} {'OUTCOME'}")
        print("-" * 60)
        for r in results:
            print(f"{r['extension']:<15} {r['status']:<10} {r.get('outcome', 'N/A')}")
        print("=" * 60)
        passed = sum(1 for r in results if r["status"] == "PASS")
        failed = len(results) - passed
        print(f"Total: {len(results)}  |  Passed: {passed}  |  Failed: {failed}")
        print()
    # --- Internal helper functions ---
    def _parse_stats(self, stats_file):
        try:
            with open(stats_file, "r") as f:
                reader = csv.DictReader(f, delimiter=";")
                rows = list(reader)
            if rows:
                last = rows[-1]
                return {
                    "successful_calls": int(last.get("SuccessfulCall", 0)),
                    "failed_calls":     int(last.get("FailedCall", 0)),
                    "response_2xx":     int(last.get("Response_2xx", 0)),
                    "response_4xx":     int(last.get("Response_4xx", 0)),
                    "response_5xx":     int(last.get("Response_5xx", 0)),
                    "retransmissions":  int(last.get("Retransmissions", 0))
                }
        except Exception as e:
            logger.warning(f"Could not parse stats: {e}")
        return {}
    def _parse_messages(self, msg_log):
        try:
            with open(msg_log, "r") as f:
                content = f.read()
            responses = re.findall(r"SIP/2\.0 (\d{3})", content)
            return {
                "call_answered":  "200" in responses,
                "busy":           "486" in responses,
                "not_found":      "404" in responses,
                "rejected":       "403" in responses,
                "auth_required":  "401" in responses,
                "all_responses":  responses
            }
        except Exception as e:
            logger.warning(f"Could not parse messages: {e}")
        return {}
    def _interpret(self, msg):
        if msg.get("call_answered"):   return "ANSWERED"
        if msg.get("busy"):            return "BUSY"
        if msg.get("not_found"):       return "NOT FOUND (404)"
        if msg.get("rejected"):        return "REJECTED (403)"
        if msg.get("auth_required"):   return "AUTH REQUIRED (401)"
        return "NO RESPONSE"

Using the Test Runner Class

Here, we have a separate file where we are importing the SIPpTestRunner class and instantiating it to execute different scenarios.

from main_class import SIPpTestRunner
if __name__ == "__main__":
    runner = SIPpTestRunner(
        scenario_file="basic_call.xml",
        target_ip="X.X.X.X",    # CUCM/SBC IP
        local_ip="X.X.X.X",     # Local Machine IP
        log_dir="./test_logs",
    )
    # Test a single extension
    result = runner.run_call("1105", duration_ms=15000, timeout_s=30)
    runner.print_summary([result])
    runner.save_results(result, "test_results.json")
    # Print a summary table
    runner.print_summary(result)
    # Save to JSON for reporting
    runner.save_results(result, "test_results.json")
    # Test multiple extensions
    extensions = ["1087", "1105", "1089", "1200", "9001", "8500"]
    results = runner.run_bulk_test(extensions, delay_between_calls=3)
    # Print a summary table
    runner.print_summary(results)
    # Save to JSON for reporting
    runner.save_results(results, "test_results.json")

Sample Terminal Output

			
============================================================
EXTENSION       STATUS     OUTCOME
------------------------------------------------------------
1087            FAIL       NOT FOUND (404)
1105            PASS       ANSWERED
1089            FAIL       NOT FOUND (404)
1200            FAIL       NOT FOUND (404)
9001            FAIL       NOT FOUND (404)
8500            FAIL       NOT FOUND (404)
============================================================
Total: 6  |  Passed: 1  |  Failed: 5

		

Sample JSON Output Data

[
  {
    "extension": "1087",
    "return_code": 1,
    "status": "FAIL",
    "successful_calls": 0,
    "failed_calls": 0,
    "response_2xx": 0,
    "response_4xx": 0,
    "response_5xx": 0,
    "retransmissions": 0,
    "call_answered": false,
    "busy": false,
    "not_found": true,
    "rejected": false,
    "auth_required": false,
    "all_responses": [
      "100",
      "404",
      "404"
    ],
    "outcome": "NOT FOUND (404)"
  },
  {
    "extension": "1105",
    "return_code": 0,
    "status": "PASS",
    "successful_calls": 0,
    "failed_calls": 0,
    "response_2xx": 0,
    "response_4xx": 0,
    "response_5xx": 0,
    "retransmissions": 0,
    "call_answered": true,
    "busy": false,
    "not_found": false,
    "rejected": false,
    "auth_required": false,
    "all_responses": [
      "100",
      "180",
      "200",
      "200",
      "200",
      "200",
      "200"
    ],
    "outcome": "ANSWERED"
  },
  {
    "extension": "1089",
    "return_code": 1,
    "status": "FAIL",
    "successful_calls": 0,
    "failed_calls": 0,
    "response_2xx": 0,
    "response_4xx": 0,
    "response_5xx": 0,
    "retransmissions": 0,
    "call_answered": false,
    "busy": false,
    "not_found": true,
    "rejected": false,
    "auth_required": false,
    "all_responses": [
      "100",
      "404",
      "404"
    ],
    "outcome": "NOT FOUND (404)"
  },
  {
    "extension": "1200",
    "return_code": 1,
    "status": "FAIL",
    "successful_calls": 0,
    "failed_calls": 0,
    "response_2xx": 0,
    "response_4xx": 0,
    "response_5xx": 0,
    "retransmissions": 0,
    "call_answered": false,
    "busy": false,
    "not_found": true,
    "rejected": false,
    "auth_required": false,
    "all_responses": [
      "100",
      "404",
      "404"
    ],
    "outcome": "NOT FOUND (404)"
  },
  {
    "extension": "9001",
    "return_code": 1,
    "status": "FAIL",
    "successful_calls": 0,
    "failed_calls": 0,
    "response_2xx": 0,
    "response_4xx": 0,
    "response_5xx": 0,
    "retransmissions": 0,
    "call_answered": false,
    "busy": false,
    "not_found": true,
    "rejected": false,
    "auth_required": false,
    "all_responses": [
      "100",
      "404",
      "404"
    ],
    "outcome": "NOT FOUND (404)"
  },
  {
    "extension": "8500",
    "return_code": 1,
    "status": "FAIL",
    "successful_calls": 0,
    "failed_calls": 0,
    "response_2xx": 0,
    "response_4xx": 0,
    "response_5xx": 0,
    "retransmissions": 0,
    "call_answered": false,
    "busy": false,
    "not_found": true,
    "rejected": false,
    "auth_required": false,
    "all_responses": [
      "100",
      "404",
      "404"
    ],
    "outcome": "NOT FOUND (404)"
  }
]

Scheduling Automated Tests

Once your main.py is working the way you want, the natural next question is – how do you run this automatically on a schedule without having to trigger it manually every time?

The simplest answer on a Linux server is cron which is a built-in job scheduler that requires no additional libraries or long-running processes. Since main.py already does the work and exits on its own, cron simply becomes the trigger that decides when to run it.

The standard format for a Cron file is given below.

			
# If you want the file to run every day at 8 AM
0 8 * * * <Path to Python executable> <Path to main.py file> >> <Path to log file for Cron output> 2>&1

To schedule it, open your crontab with crontab -e and add a single line as mentioned above.

Important : One important thing to keep in mind is that cron runs in a minimal environment and doesn’t know where your project folder is. So make sure all file paths inside main.py file including the scenario file, audio file, log directory etc are absolute paths rather than relative ones. It is also a good idea to redirect the output to a log file by appending >> /path/to/cron.log 2>&1 at the end of the cron entry. So, you have a record of every run and can debug easily if something goes wrong.

YouTube Explainer

I am currently working on the video where I go through this whole process in detail. It should be live in the next couple of days. Stay tuned!!

In the meantime, if you want, you can check out part 1 of this series here which I shared in my previous blog.

https://www.youtube.com/watch?v=inulTQSeWwQ

What’s Next ?

At this point you have everything you need to run SIPp from Python, test extensions in bulk, parse the results and save them to JSON. But not everyone you work with is comfortable running Python scripts from a terminal, and in a real-world scenario you’d want something a bit more polished that your team or even non-technical stakeholders can actually use.

In the next part of this series, I’ll show you a demo of a product with a front end GUI that I recently developed for one of our customers. It runs on top of almost everything we have covered so far. This gives us a proper interface to configure our tests, kick them off with a click, monitor results in real time and export reports without touching a terminal window. This is where the solution starts looking less like a standalone script and more like an actual product.

Stay tuned for that and as always, feel free to drop your questions or feedback in the comments below. Until then, Keep Learning!!

Let’s connect on LinkedIn