Testing

Telecom Billing System Testing: Strategies for CDR, Rating, and Invoice Accuracy

HelpMeTest

HelpMeTest

— 6 min read
Telecom Billing System Testing: Strategies for CDR, Rating, and Invoice Accuracy

Telecom billing is where network events become revenue. A single rating error silently applied to millions of CDRs (Call Detail Records) can cost or lose millions of dollars before discovery. Billing system testing is not glamorous, but it is one of the highest-impact testing disciplines in the industry. Errors that make it to invoices generate customer complaints, regulatory scrutiny, and churn.

Billing System Architecture

A telecom billing system typically consists of these pipeline stages:

Network Events → Mediation → Rating Engine → Billing → Invoicing
     │               │             │            │          │
  Switches         CDR gen      Tariff calc   Account   PDF/EDI
  PCRF/OCS         Normalization  Discount     balance   generation
  IMS nodes        Dedup          Tax           cycle
                   Aggregation    Bundling

Each stage has distinct test concerns:

  • Mediation: Are all events captured? Are duplicates correctly suppressed? Is normalization lossless?
  • Rating: Is the correct tariff applied? Are discounts, bundles, and promotions applied correctly?
  • Billing: Do account balances update correctly? Are credit limits enforced?
  • Invoicing: Does the invoice math add up? Are taxes correct by jurisdiction?

CDR (Call Detail Record) Testing

CDRs are the atomic unit of billing. A voice CDR contains: calling/called number, start time, duration, network node, quality indicators. A data CDR contains: MSISDN, IMSI, bytes up/down, session start/end, APN.

CDR Generation Validation

import pandas as pd
from datetime import datetime, timedelta
import random

def generate_test_cdrs(count=1000, output_file="test_cdrs.csv"):
    """Generate synthetic CDRs for testing mediation and rating."""
    cdrs = []
    base_time = datetime(2024, 1, 15, 8, 0, 0)
    
    msisdns = [f"447{random.randint(700000000, 799999999)}" for _ in range(50)]
    
    for i in range(count):
        call_type = random.choice(["voice", "sms", "data"])
        start = base_time + timedelta(seconds=random.randint(0, 86400))
        
        if call_type == "voice":
            duration = random.randint(5, 3600)
            cdr = {
                "record_type": "MOC",
                "msisdn": random.choice(msisdns),
                "called_number": f"+3531{random.randint(1000000, 9999999)}",
                "start_time": start.isoformat(),
                "duration_seconds": duration,
                "network_type": random.choice(["2G", "3G", "4G", "5G"]),
                "roaming": random.choice([0, 0, 0, 1]),  # 25% roaming
                "bytes_uplink": 0,
                "bytes_downlink": 0,
            }
        elif call_type == "data":
            duration = random.randint(60, 7200)
            bytes_dl = random.randint(1024, 100 * 1024 * 1024)
            cdr = {
                "record_type": "GPRS",
                "msisdn": random.choice(msisdns),
                "called_number": "",
                "start_time": start.isoformat(),
                "duration_seconds": duration,
                "network_type": random.choice(["4G", "5G"]),
                "roaming": random.choice([0, 0, 0, 1]),
                "bytes_uplink": int(bytes_dl * 0.1),
                "bytes_downlink": bytes_dl,
            }
        else:  # SMS
            cdr = {
                "record_type": "SMS-MO",
                "msisdn": random.choice(msisdns),
                "called_number": f"+3531{random.randint(1000000, 9999999)}",
                "start_time": start.isoformat(),
                "duration_seconds": 0,
                "network_type": "4G",
                "roaming": 0,
                "bytes_uplink": 0,
                "bytes_downlink": 0,
            }
        cdrs.append(cdr)
    
    df = pd.DataFrame(cdrs)
    df.to_csv(output_file, index=False)
    return df

def test_cdr_deduplication(mediation_input, mediation_output):
    """Verify mediation layer correctly deduplicates CDRs."""
    input_df = pd.read_csv(mediation_input)
    output_df = pd.read_csv(mediation_output)
    
    # Create duplicate CDRs in input (simulating network switch retry)
    duplicates = input_df.sample(n=50)
    input_with_dups = pd.concat([input_df, duplicates])
    input_with_dups.to_csv(mediation_input, index=False)
    
    # Re-run mediation (system under test)
    run_mediation(mediation_input, mediation_output)
    
    output_df = pd.read_csv(mediation_output)
    
    # Output should have same count as original input (no duplicates)
    expected_count = len(input_df)
    actual_count = len(output_df)
    
    assert actual_count == expected_count, \
        f"Deduplication failed: input={expected_count}, output={actual_count} " \
        f"(expected no duplicates)"
    print(f"PASS: Deduplication correct — {actual_count} records after removing {len(duplicates)} duplicates")

Mediation Normalization Testing

def test_duration_normalization():
    """Billing durations must be normalized to billing unit (typically 60s for voice)."""
    test_cases = [
        # (actual_duration_s, billing_unit_s, expected_billed_units)
        (45, 60, 1),    # Less than 1 unit → still billed 1 unit (rounding up)
        (60, 60, 1),    # Exactly 1 unit → 1 unit
        (61, 60, 2),    # 1 second over → 2 units
        (120, 60, 2),   # Exactly 2 units → 2 units
        (0, 60, 0),     # Zero duration (failed call) → 0 units
    ]
    
    for actual, unit, expected in test_cases:
        billed = math.ceil(actual / unit) if actual > 0 else 0
        assert billed == expected, \
            f"Duration {actual}s with {unit}s billing unit: expected {expected} units, got {billed}"
    
    print("PASS: Duration normalization correct for all test cases")

Rating Engine Accuracy Testing

The rating engine applies tariffs to normalized CDRs. This is where the most financially significant errors occur.

from decimal import Decimal, ROUND_HALF_UP
import json

def load_tariff(tariff_file):
    with open(tariff_file) as f:
        return json.load(f)

def calculate_expected_charge(cdr, tariff):
    """
    Reference rating implementation for test validation.
    Compare this against the production rating engine output.
    """
    charge = Decimal("0.00")
    
    if cdr["record_type"] == "MOC":
        # Voice call rating
        if cdr["roaming"]:
            rate = Decimal(str(tariff["roaming"]["voice_per_minute"]))
        else:
            # Check if within free bundle allowance
            if cdr.get("bundle_minutes_remaining", 0) > 0:
                rate = Decimal("0.00")
            else:
                rate = Decimal(str(tariff["domestic"]["voice_per_minute"]))
        
        billed_minutes = (cdr["duration_seconds"] + 59) // 60  # Round up
        charge = rate * Decimal(str(billed_minutes))
    
    elif cdr["record_type"] == "GPRS":
        # Data rating: typically per MB
        bytes_total = cdr["bytes_uplink"] + cdr["bytes_downlink"]
        mb_consumed = Decimal(str(bytes_total)) / Decimal("1048576")
        
        if cdr["roaming"]:
            rate = Decimal(str(tariff["roaming"]["data_per_mb"]))
            charge = (mb_consumed * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
        else:
            # Domestic data may be free within bundle
            charge = Decimal("0.00")
    
    elif cdr["record_type"] == "SMS-MO":
        if cdr["roaming"]:
            charge = Decimal(str(tariff["roaming"]["sms_per_message"]))
        else:
            charge = Decimal("0.00")  # Assuming free domestic SMS in bundle
    
    return charge

def verify_rating_accuracy(rated_cdrs_file, tariff_file, tolerance=0.001):
    """
    Compare production rating engine output against reference implementation.
    tolerance: acceptable deviation as fraction (0.001 = 0.1%)
    """
    tariff = load_tariff(tariff_file)
    rated_df = pd.read_csv(rated_cdrs_file)
    
    mismatches = []
    
    for _, row in rated_df.iterrows():
        cdr = row.to_dict()
        expected = calculate_expected_charge(cdr, tariff)
        actual = Decimal(str(cdr["charge"]))
        
        if expected == 0 and actual == 0:
            continue
        
        if expected > 0:
            deviation = abs(actual - expected) / expected
        else:
            deviation = abs(actual - expected)
        
        if deviation > Decimal(str(tolerance)):
            mismatches.append({
                "cdr_id": cdr.get("cdr_id"),
                "type": cdr["record_type"],
                "expected": float(expected),
                "actual": float(actual),
                "deviation_pct": float(deviation * 100)
            })
    
    if mismatches:
        print(f"RATING MISMATCHES: {len(mismatches)}/{len(rated_df)}")
        for m in mismatches[:10]:
            print(f"  CDR {m['cdr_id']} ({m['type']}): expected={m['expected']:.4f}, "
                  f"actual={m['actual']:.4f} ({m['deviation_pct']:.2f}% deviation)")
        return False
    
    print(f"PASS: All {len(rated_df)} CDRs rated correctly within {tolerance*100:.1f}% tolerance")
    return True

Invoice Reconciliation Testing

Invoices must be mathematically correct and reconcilable back to CDRs.

def reconcile_invoice(invoice_file, rated_cdrs_file, account_id):
    """Verify invoice total matches sum of rated CDRs for the billing period."""
    
    invoice = parse_invoice(invoice_file)  # Returns {account_id, period, line_items, total, tax, grand_total}
    cdrs = pd.read_csv(rated_cdrs_file)
    account_cdrs = cdrs[cdrs["account_id"] == account_id]
    
    period_start = pd.Timestamp(invoice["period_start"])
    period_end = pd.Timestamp(invoice["period_end"])
    
    period_cdrs = account_cdrs[
        (pd.to_datetime(account_cdrs["start_time"]) >= period_start) &
        (pd.to_datetime(account_cdrs["start_time"]) < period_end)
    ]
    
    cdr_total = period_cdrs["charge"].sum()
    invoice_charges = sum(item["amount"] for item in invoice["line_items"])
    
    # Allow 1 cent rounding tolerance
    discrepancy = abs(cdr_total - invoice_charges)
    assert discrepancy < 0.01, \
        f"Invoice discrepancy: CDR total={cdr_total:.2f}, " \
        f"Invoice charges={invoice_charges:.2f}, " \
        f"Difference={discrepancy:.4f}"
    
    # Verify tax calculation
    expected_tax = invoice_charges * get_tax_rate(invoice["jurisdiction"])
    tax_discrepancy = abs(invoice["tax"] - expected_tax)
    assert tax_discrepancy < 0.01, \
        f"Tax calculation error: expected={expected_tax:.2f}, actual={invoice['tax']:.2f}"
    
    print(f"PASS: Invoice reconciled — {len(period_cdrs)} CDRs, total={invoice_charges:.2f}")

Roaming Billing Testing (TAP Files)

Roaming billing uses TAP (Transferred Account Procedures) files exchanged between operators. Testing requires generating synthetic TAP files and verifying they are correctly processed.

def generate_tap3_file(cdrs, home_operator_tadig, visited_operator_tadig, output_file):
    """
    Generate a TAP3 (GSM Association TD.57) file from CDR data.
    TAP3 is ASN.1-encoded; this example uses the tap3-python library.
    """
    from tap3 import TAP3File, MobileOrigCall, BasicCallInformation
    from datetime import datetime
    
    tap = TAP3File()
    tap.sender_identification = visited_operator_tadig
    tap.recipient_identification = home_operator_tadig
    tap.file_creation_timestamp = datetime.utcnow()
    
    for cdr in cdrs:
        if not cdr["roaming"]:
            continue
        
        call = MobileOrigCall()
        call.calling_number = cdr["msisdn"]
        call.called_number = cdr["called_number"]
        call.call_event_start_timestamp = cdr["start_time"]
        call.total_call_event_duration = cdr["duration_seconds"]
        call.charge_information.charge = int(float(cdr["charge"]) * 100)  # In cents
        call.charge_information.currency = "EUR"
        tap.call_events.append(call)
    
    tap.write(output_file)
    print(f"Generated TAP3 file: {len(tap.call_events)} roaming CDRs → {output_file}")

def test_tap_file_processing(tap_file, expected_roaming_revenue):
    """Verify TAP file is correctly ingested and revenue matches."""
    # Submit TAP file to billing system (system-specific API)
    submit_tap_file(tap_file)
    
    # Wait for processing (may take minutes in production)
    import time
    time.sleep(30)
    
    # Verify ingested revenue
    processed_revenue = get_roaming_revenue_for_period(
        operator="TEST_VISITED",
        period_start="2024-01-15",
        period_end="2024-01-16"
    )
    
    tolerance = expected_roaming_revenue * 0.001  # 0.1% tolerance
    assert abs(processed_revenue - expected_roaming_revenue) < tolerance, \
        f"Roaming revenue mismatch: expected={expected_roaming_revenue:.2f}, " \
        f"processed={processed_revenue:.2f}"
    print(f"PASS: TAP file processed correctly — roaming revenue={processed_revenue:.2f}")

Performance Testing for Peak CDR Volumes

Billing systems must handle end-of-month peaks without falling behind real-time.

# CDR throughput test using Apache Kafka as mediation input
# Test: can rating engine process 10,000 CDRs/second?

# Generate test CDRs at target rate
python3 generate_cdrs.py --count 3600000 --output test_cdrs.csv  # 1 hour of CDRs

# Push to Kafka at controlled rate
python3 << 'EOF'
from kafka import KafkaProducer
import csv
import time
import json

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode()
)

TARGET_RATE = 10000  # CDRs per second
batch_size = 1000
interval = batch_size / TARGET_RATE

with open("test_cdrs.csv") as f:
    reader = csv.DictReader(f)
    batch_start = time.monotonic()
    count = 0
    
    for row in reader:
        producer.send("cdrs", value=row)
        count += 1
        
        if count % batch_size == 0:
            producer.flush()
            elapsed = time.monotonic() - batch_start
            sleep_time = interval - elapsed
            if sleep_time > 0:
                time.sleep(sleep_time)
            batch_start = time.monotonic()
            print(f"Sent {count} CDRs, rate: {batch_size/max(elapsed,0.001):.0f}/s")

producer.flush()
print(f"Completed: {count} CDRs sent")
EOF

# Monitor rating engine consumer lag
watch -n 2 "kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group billing-rating-engine | grep cdr"

Key Takeaways

  • CDR deduplication testing is critical — network switches often retry event records, and duplicate charges are a common regulatory complaint
  • Build a reference rating implementation and run it against production outputs; financial precision requires Decimal arithmetic, not floating point
  • Invoice reconciliation must trace from every line item back to individual CDRs — a reconciliation gap of even 0.01% at scale is a significant revenue leak
  • TAP file testing for roaming requires generating syntactically valid ASN.1-encoded files; malformed TAP files are silently rejected by most inter-operator clearing houses
  • Performance test at 2-3x peak CDR volume — billing systems that fall behind real-time during month-end peaks create delayed invoices and customer experience problems

Read more

Start now free