Swift Network Testing - Automate XCTest with Python

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Usage:
#  $ ./test_network.py urls
#  $ ./test_network.py < urls
#
# Import needed stdlib and other dependencies
#
from __future__ import print_function
import os, sys, time, subprocess
 
def cutOutNextOccurrenceOfString(std, needle, endKeyWord):
    v = std.find(needle)
    if v != -1:
        rtn_str = ""
        let = (v + 1000)
        while v < len:
            char = std[v]
            if char is endKeyWord[0]:
                next_char = endKeyWord[1]
                next_next_char = endKeyWord[2]
                next_std = std[(v+1)]
                next_next_std = std[(v+2)]
                # Consider a three char match good enough
                if next_std is next_char and next_next_std is next_next_char:
                    break
            rtn_str += char
            v += 1
        std = std[v:]
        return (1, std, rtn_str)
    else:
        return (-1, std, "")
 
# Main function
def main():
    # Create some local variables
    file_input_name = ""  # Input name for the file being passed in by the command line
    complete_content = "" # The string contents of the file
    file_lines = [] # A list of lines that the complete_content will be exploded to
    file_input_stdin = None # File from stdin, i.e., < input.txt
 
    print("----------------- Starting Program -----------------")
    # ---------------- Parse the command line input ----------------
 
    # Use sys.argv or sys.stdin to get the file input from the command line
    if len(sys.argv) > 1 and sys.argv[1] is not None:
        # Recognize that sys.argv has a file argument in it
        file_input_name = sys.argv[1]
    elif not sys.stdin.isatty():
        # isatty() makes sure stdin has data http://man7.org/linux/man-pages/man3/isatty.3.html
        # Recognize that there is a file object in sys.stdin
        file_input_stdin = sys.stdin
    else:
        # Write the raised error
        error_raised = "OOPs! It seems that there was an issue locating your input file"
        # Display the error to the console
        print(error_raised)
        exit("Please provide a valid filename or file as a command line input argument")
 
 
    # Check whether the file_input_name from sys.argv is actually a valid file
    if file_input_stdin is None:
        if not os.path.isfile(file_input_name):
            # Write the raised error
            # Display the error to the console
            print("OOPs! It seems there was an issue locating your filename")
            exit("Please provide a valid filename")
 
        # ---------------- Attempt to open the file ----------------
 
        # Open file, read contents into a list to be parsed
        try:
            # Open the file and read the contents into a
            if file_input_stdin is None:
                with open(file_input_name, 'r') as file:
                    complete_content = file.read()
            else:
                complete_content = file_input_stdin.read()
 
            # Explode or split the string contents to a list
            file_lines = complete_content.split("\n")
 
        except ValueError:
            # Write the raised error
            error_raised = "Unexpected error loading file: " + str(sys.exc_info()[0])
            # Display the error to the console
            print(error_raised)
            exit("This program needs an input file to continue. Exiting...")
 
        # ---------------- Parse the file ----------------
        urls = ""
        i = 0
        url_list = []
        log = ""
 
        for url in file_lines:
            i += 1
            print(url)
            urls += url + ","
 
            if i % 2 == 0:
                if urls.endswith(","):
                    urls = urls[:-1]
 
                test_command = "xcodebuild \
                                -project '../Agnostic Development.xcodeproj' \
                                -scheme AgnosticDevelopmentLatencyTest \
                                -destination 'platform=iOS,name=Matts iPhone 7+' \
                                URL_ARGS=" + urls +" \
                                clean test"
 
                print("Working...")
                stdout_response = subprocess.check_output(test_command,stderr=subprocess.STDOUT,shell=True)
                flag = 1
                while flag is 1:
                    tuple = cutOutNextOccurrenceOfString(stdout_response, "SUCCESS", "|DONE|")
                    flag = tuple[0]
                    stdout_response = tuple[1]
                    log += tuple[2] + "\n"
 
                urls = ""
        print(log)
        print("----------------- Ending Program -----------------")
        exit()
    else:
        # Write the raised error
        error_raised = "OOPs! It seems that there was an issue locating your input file"
 
# Execute the main function
if __name__ == '__main__':
    main()

class Network: NSObject {
 
    //
    // MARK: - Shared instance property
    //
    public static var shared = Network()
 
    //
    // MARK: - Public instance properties
    //
    public weak var networkDelegate: NetworkProtocol?
 
    //
    // MARK: - Private constants
    //
    private let url = "https://5fc3d7589074cd0c4bf5-79ef711e857aec8d77eb74e0027f6262.ssl.cf1.rackcdn.com/articles.json"
    private let urlConfiguration = URLSessionConfiguration.default
 
    //
    // MARK: - Public Instance Methods
    //
 
    // Load articles over the network with a default completion block provided by URLSession
    public func loadNewArticlesWithDefaultCompletion() {
 
        guard let articleURL = URL(string: url) else {
            return
        }
 
        var articleRequest = URLRequest(url: articleURL)
        articleRequest.httpMethod = "GET"
 
        let urlSession = URLSession(configuration: urlConfiguration)
 
        // Create a dataTask with a closure that defines the comletion handler
        // The closure in this case is defined as completionHandler: { (data, response, error) in  ... }
        let articleTask = urlSession.dataTask(with: articleRequest, completionHandler: { [weak self] (data, response, error) in
 
            // Ensure that an error is not present, otherwise, return the error
            guard error == nil else {
                // Usage of weak self to reference networkDelegate
                DispatchQueue.main.async {
                    self?.networkDelegate?.networkReceivedError(error: error.debugDescription)
                }
                return
            }
 
            // Ensure that the network data is available and that the byte count is greater that zero
            guard let networkData = data, networkData.count > 0 else {
                DispatchQueue.main.async {
                    self?.networkDelegate?.networkReceivedError(error: "There was an error parsing network data.")
                }
                return
            }
 
            // Perform the JSONSerialization into an array of Dictionaries
            // Next use the convenience constructor in Article to create an Article object out of each object in the JSON array
            do {
                if let networkArticles = try JSONSerialization.jsonObject(with: networkData, options: []) as? [[String: AnyObject]] {
                    var articles: [Article] = []
 
                    // Create article objects out of a JSON object
                    for networkArticle in networkArticles {
                        let article = Article(jsonObject: networkArticle)
                        articles.append(article)
                    }
                    DispatchQueue.main.async {
                        self?.networkDelegate?.articlesFinishedLoading(articles: articles)
                    }
                }
            } catch let error as NSError {
                DispatchQueue.main.async {
                    self?.networkDelegate?.networkReceivedError(error: error.debugDescription)
                }
            }
 
        })
        articleTask.resume()
    }
 
    // Load articles over the network with a assigned completion as a passed in argument : networkArticlesCompletionHander
    // networkArticlesCompletionHander is a closure defined as a variable to handle all of the processing without using URLSessions out of the box completion handler
    public func loadNewArticlesWithAssignedCompletion(networkArticlesCompletionHander: @escaping(Data?, URLResponse?, Error?) -> Void) {
 
        guard let articleURL = URL(string: url) else {
            return
        }
 
        var articleRequest = URLRequest(url: articleURL)
        articleRequest.httpMethod = "GET"
 
        let urlSession = URLSession(configuration: urlConfiguration)
 
        let articleTask = urlSession.dataTask(with: articleRequest, completionHandler: networkArticlesCompletionHander)
        articleTask.resume()
    }
 
    public func runNetworkingRequests(networkRequest: URLRequest, networkSession: URLSession, completion: @escaping (_ : String) -> Void) {
        var results: String = ""
        let start = Date()
 
        // Actual network request
        let networkTask = networkSession.dataTask(with: networkRequest, completionHandler: { (data, response, error) in
 
            guard let url = response?.url else {
                return
            }
            if let _ = data {
                let elapsed_start = abs(start.timeIntervalSinceNow)
                results = "SUCCESS: Latency Elapsed: \(elapsed_start) for URL: \(url) |DONE|"
                DispatchQueue.main.async {
                    print(results)
                    completion(results)
                }
            } else {
                let elapsed_fail = abs(start.timeIntervalSinceNow)
                results = "FAIL: Latency Elapsed: \(elapsed_fail) for URL: \(url) |DONE|"
                DispatchQueue.main.async {
                    print(results)
                    completion(results)
                }
            }
        })
        networkTask.resume()
    }
}

import XCTest
@testable import Agnostic_Development
 
class AgnosticDevelopmentLatencyTest: XCTestCase {
    //
    // MARK: Public Constants
    //
    public let urlConfiguration: URLSessionConfiguration = URLSessionConfiguration.default
    //
    // MARK: Public Instance Properties
    //
    public var urls: [String] = []
    public var networkRequests: [URLRequest] = []
    public var networkSessions: [URLSession] = []
    public var results: [String] = []
 
    override func setUp() {
        super.setUp()
 
        // Parse the comma separated TEST_ARGs variable
        if let TEST_ARGS_URLS = ProcessInfo.processInfo.environment["URL_ARGS"] {
            urls = TEST_ARGS_URLS.components(separatedBy: ",")
            NSLog("Launch Data URL Count: \(urls.count)")
        } else {
            // If you do not get any test args, assert failure
            XCTAssert(false, "FAIL: No data found in TEST_ARGS_URLS")
        }
 
        // Put setup code here. This method is called before the invocation of each test method in the class.
        for networkURL in urls {
            guard let networkURLObj = URL(string: networkURL) else {
                return
            }
            var networkRequest = URLRequest(url: networkURLObj)
            networkRequest.httpMethod = "GET"
            networkRequests.append(networkRequest)
 
            let networkSession = URLSession(configuration: urlConfiguration)
            networkSessions.append(networkSession)
        }
    }
 
    override func tearDown() {
        // Put teardown code here. This method is called after the invocation of each test method in the class.
        super.tearDown()
 
    }
 
    ...
 
}

func testFirstNetworkRequestLatency() {
 
    guard let url = networkRequests[0].url,
        networkRequests.count > 0 else {
        return
    }
    let expectation = XCTestExpectation(description: "Download from: \(url)")
    Network.shared.runNetworkingRequests(networkRequest: networkRequests[0], networkSession: networkSessions[0]) {
        (result: String) in
        if result.range(of:"FAIL") != nil {
            XCTAssert(false, "URLDebug: A failing message was found on the first execution of the request. \(result)")
        } else {
            XCTAssert(true, "URLDebug: The first execution was run successfully. \(result)")
        }
        expectation.fulfill()
    }
 
    wait(for: [expectation], timeout: 15.0)
}

$ ./test_network.py urls