#!/usr/bin/python2
# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Unit tests for apache_access_log_metrics.py"""
from __future__ import print_function
import os
import mock
import re
import subprocess
import tempfile
import unittest
import apache_access_log_metrics
SCRIPT_PATH = os.path.abspath(
os.path.join(os.path.dirname(__file__),
'apache_access_log_metrics.py'))
EXAMPLE_REQUEST_LINE = (
r'chromeos-server2.mtv.corp.google.com:80 100.108.96.5 - - '
r'[19/May/2017:11:47:03 -0700] '
r'"POST /afe/server/noauth/rpc/?method=foo HTTP/1.1\"" '
r'200 354 "-" "Python-urllib/2.7" 5'
)
class TestParsers(unittest.TestCase):
"""Tests the parsing functions in apache_access_log_metrics."""
def testParseResponse(self):
"""Tests that the regex matches the example log line."""
match = apache_access_log_metrics.ACCESS_MATCHER.match(
EXAMPLE_REQUEST_LINE)
self.assertTrue(match)
self.assertEqual(match.group('bytes_sent'), '354')
self.assertEqual(match.group('response_seconds'), '5')
class TestEmitters(unittest.TestCase):
"""Tests the emitter functions in apache_access_log_metrics."""
def testEmitResponse(self):
"""Tests that the matcher function doesn't throw an Exception."""
match = apache_access_log_metrics.ACCESS_MATCHER.match(
EXAMPLE_REQUEST_LINE)
# Calling the emitter should not raise any exceptions (for example, by
# referencing regex match groups that don't exist.
with mock.patch.object(apache_access_log_metrics, 'metrics'):
apache_access_log_metrics.EmitRequestMetrics(match)
class TestScript(unittest.TestCase):
"""Tests the script end-to-end."""
def testApachAccessLogScriptWithMatchingLine(self):
"""Try shelling out the the script with --debug-file.
Sending it a line which matches the first-line regex should result in
output from ACCESS_TIME_METRIC.
"""
with tempfile.NamedTemporaryFile() as temp_file:
p = subprocess.Popen([SCRIPT_PATH,
'--debug-metrics-file', temp_file.name],
stdin=subprocess.PIPE, stdout=subprocess.PIPE)
p.communicate(EXAMPLE_REQUEST_LINE)
with open(temp_file.name) as fh:
contents = fh.read()
self.assertTrue(re.search(
apache_access_log_metrics.ACCESS_TIME_METRIC[1:] + r'\b',
contents))
self.assertTrue(re.search(
apache_access_log_metrics.ACCESS_BYTES_METRIC[1:] + r'\b',
contents))
if __name__ == '__main__':
unittest.main()