CMPUT 404

'''
More than one item should exist for clustering to take place. So if we only have one DataEntry in
a timeline/trace, then it just goes in its own cluster and we skip the clustering process. 
'''
if(len(all_embeddings) > 1):
    <clustering logic>
else:
    <skip clustering>

'''
all_embeddings will contain an n by m vector where n is the number of dimensions for the embeddings, and m is the number of embeddings (or timeline entities of this symbol). 

So each 'row' in this variable is an embedding of a timeline entity for this symbol.

This is the format expected by do_h_clustering(). 
'''
all_embeddings = np.vstack(_embeddings)

'''
All of a premium user's likes are converted into super likes before the like view is displayed.
The html template will highlight elements in super_likes. While elements in user_likes are treated normally.
'''
for like_object in user_likes: 
    super_likes.append(SuperLike(like_object))    

    def read_row(self, row): # this is called once for every row in the menti export
        # Usually each student has their own row, but if they reconnect they might have several
        name = self.find_name(row) 
        if name is None: # returned when find_name can't find any codenames in the row
            return 
        self.add_name(name) # set up some data structures before we process the questions and answers
        responded = {question: None for question, _ in self.question_response_count.items()} # we need a new dictionary for every student (row), but we'll fill in the values later

# If the length of the all embeddings variable is bigger than 1
if(len(all_embeddings) > 1):
    <clustering logic>
else:
    <skip clustering>

# Iterate throught all like objects in user_likes. 
for like_object in user_likes: 
    # Create a super like and append it to super_likes
    super_likes.append(SuperLike(like_object))    

# Apply the vstack function from numpy  (https://numpy.org/) to the _embeddings and store the result in all_embeddings.
all_embeddings = np.vstack(_embeddings)

    def read_row(self, row): # reads the row
        name = self.find_name(row) # find the name
        if name is None: # if there's no name
            return # give up
        self.add_name(name) # add the name
        responded = {question: None for question, _ in self.question_response_count.items()} # make a dictionary of the questions

import asyncio
import socket
import random

AI_ADDR_INDEX = 4
ADDR_HOST_INDEX = 0

class Hostnames(dict):
    def __init__(self):
        super().__init__()
        self.tasks = {} 
        self.started = {}
        self.times = {}

    async def get_hostname(self, hostname):
        loop = asyncio.get_running_loop()
        async def resolve():
            self.started[hostname] = loop.time()
            try:
                addrinfos = await loop.getaddrinfo(hostname, None, family=socket.AF_INET)
            except (socket.gaierror, asyncio.TimeoutError):
                ip = None
            else:
                ip = addrinfos[0][AI_ADDR_INDEX][ADDR_HOST_INDEX]
            self.times[hostname] = loop.time() - self.started[hostname]
            return ip
        if hostname in self:
            return self[hostname]
        else:
            if hostname in self.tasks:
                task = self.tasks[hostname]
            else:
                task = loop.create_task(resolve())
                self.tasks[hostname] = task

            assert self.tasks[hostname] is task, f"Task for {hostname} not found"
            assert isinstance(task, asyncio.Task), f"Expected Task, got {type(task)} for {hostname}"

            try:
                ip = await task
            except asyncio.CancelledError:
                if hostname in self.tasks:
                    del self.tasks[hostname]
                return None
            else:
                assert isinstance(ip, str) or ip is None, f"Expected IP address, got {ip} for {hostname}"
                self[hostname] = ip
                return ip

    def start_get_hostname(self, hostname):
        loop = asyncio.get_running_loop()
        return loop.create_task(self.get_hostname(hostname))

    def __getitem__(self, key):
        if super().__contains__(key):
            return super().__getitem__(key)
        else:
            self.start_get_hostname(key)
            return None

async def main():
    hostnames = Hostnames()
    def generate_random_hostname():
        return ''.join(random.choices('abcdefghijklmnopqrstuvwxyz', k=4)) + ".com"
    names = [generate_random_hostname() for _ in range(10)]
    for name in names:
        hostnames[name]
    assert hostnames["example.com"] is None, "Should not have resolved yet"
    await asyncio.sleep(0.2)
    for name in names:
        ip = hostnames[name]
        time = hostnames.times.get(name, float('inf'))
        print(f"{name} -> {ip}, time: {time:.3f}s")

if __name__ == "__main__":
    asyncio.run(main())

import unittest
from unittest.mock import patch, AsyncMock, MagicMock
import asyncio
import socket
from hostnames import Hostnames

class TestHostnames(unittest.IsolatedAsyncioTestCase):
    async def test_get_hostname_success(self):
        hostnames = Hostnames()
        mock_ip = "192.168.1.1"
        real_loop = asyncio.get_running_loop()
        with patch("asyncio.get_running_loop") as mock_get_loop:
            mock_loop = MagicMock(wraps=real_loop)
            mock_get_loop.return_value = mock_loop
            mock_loop.getaddrinfo = AsyncMock(return_value=[
                (None, None, None, None, (mock_ip,))
            ])
            ip = await hostnames.get_hostname("example.com")
            self.assertEqual(ip, mock_ip)
            self.assertIn("example.com", hostnames)
            self.assertEqual(hostnames["example.com"], mock_ip)
            self.assertIn("example.com", hostnames.tasks)
            self.assertIn("example.com", hostnames.times)
            self.assertIn("example.com", hostnames.started)
            self.assertIsInstance(hostnames.tasks["example.com"], asyncio.Task)
            mock_loop.getaddrinfo.assert_awaited_once_with("example.com", None, family=socket.AF_INET)

    async def test_get_hostname_failure(self):
        hostnames = Hostnames()
        real_loop = asyncio.get_running_loop()
        with patch("asyncio.get_running_loop") as mock_get_loop:
            mock_get_loop.return_value = MagicMock(wraps=real_loop)
            mock_get_loop.return_value.getaddrinfo = AsyncMock(side_effect=socket.gaierror)
            ip = await hostnames.get_hostname("invalid.com")
            self.assertIsNone(ip)
            self.assertIs(hostnames["invalid.com"], None)

    async def test_start_get_hostname(self):
        hostnames = Hostnames()
        with patch.object(hostnames, "get_hostname", new_callable=AsyncMock) as mock_get_hostname:
            task = hostnames.start_get_hostname("example.com")
            self.assertIsInstance(task, asyncio.Task)
            await task
            mock_get_hostname.assert_awaited_once_with("example.com")

    async def test_getitem_triggers_start_get_hostname(self):
        hostnames = Hostnames()
        with patch.object(hostnames, "start_get_hostname", return_value=None) as mock_start:
            self.assertIsNone(hostnames["example.com"])
            mock_start.assert_called_once_with("example.com")

    async def test_getitem_triggers_get_hostname(self):
        hostnames = Hostnames()
        with patch.object(hostnames, "get_hostname", return_value=None, new_callable=AsyncMock) as mock_get_hostname:
            self.assertIsNone(hostnames["example.com"])
            mock_get_hostname.assert_called_once_with("example.com")

    async def test_get_hostname_success2(self):
        hostnames = Hostnames()
        mock_ip = "192.168.1.1"
        real_loop = asyncio.get_running_loop()
        with patch("asyncio.get_running_loop") as mock_get_loop:
            mock_loop = MagicMock(wraps=real_loop)
            mock_get_loop.return_value = mock_loop
            gai = AsyncMock(return_value=[
                (None, None, None, None, (mock_ip,))
            ])
            mock_loop.getaddrinfo = gai
            assert hostnames["example.com"] is None
            while hostnames["example.com"] is None:
                await asyncio.sleep(0)
            ip = hostnames["example.com"]
            self.assertEqual(ip, mock_ip)
            self.assertIn("example.com", hostnames)
            self.assertEqual(hostnames["example.com"], mock_ip)
            self.assertIn("example.com", hostnames.tasks)
            self.assertIn("example.com", hostnames.times)
            self.assertIn("example.com", hostnames.started)
            self.assertIsInstance(hostnames.tasks["example.com"], asyncio.Task)
            gai.assert_awaited_once_with("example.com", None, family=socket.AF_INET)



if __name__ == "__main__":
    unittest.main()

import heapq  # Do stuff with this

def dijkstra(graph, start):  # Start Dijkstra here
    distances = {node: float('inf') for node in graph}  # Make everything big
    distances[start] = 0  # Zero this one
    queue = [(0, start)]  # First thing in line

    while queue:  # While still stuff
        curr_distance, curr_node = heapq.heappop(queue)  # Get next thing

        if curr_distance > distances[curr_node]:  # Skip if it's worse
            continue  # Skip over this

        for neighbor, weight in graph[curr_node]:  # Check things next to it
            distance = curr_distance + weight  # Add up some stuff
            if distance < distances[neighbor]:  # If it's less, do it
                distances[neighbor] = distance  # Set the new one
                heapq.heappush(queue, (distance, neighbor))  # Put it in there
    return distances  # Give it back

import heapq  # Import heap queue for priority handling

def dijkstra(graph, start):  # Dijkstra's algorithm main function
    distances = {node: float('inf') for node in graph}  # Initialize all distances to infinity
    distances[start] = 0  # Set start node distance to zero
    queue = [(0, start)]  # Priority queue with starting node

    while queue:  # Process until queue is empty
        curr_distance, curr_node = heapq.heappop(queue)  # Pop the node with smallest distance

        if curr_distance > distances[curr_node]:  # Ignore if outdated distance
            continue  # Skip to next loop iteration

        for neighbor, weight in graph[curr_node]:  # Iterate over neighbors
            distance = curr_distance + weight  # Calculate new tentative distance
            if distance < distances[neighbor]:  # If shorter path found
                distances[neighbor] = distance  # Update shortest distance
                heapq.heappush(queue, (distance, neighbor))  # Push updated node to queue
    return distances  # Return final shortest paths

import heapq  # Use heap for efficiently retrieving the closest unvisited node

def dijkstra(graph, start):  # Compute shortest paths from a start node using Dijkstra’s algorithm
    distances = {node: float('inf') for node in graph}  # Start with infinite distance to all nodes
    distances[start] = 0  # Distance to the start node is zero
    queue = [(0, start)]  # Initialize the priority queue with the start node

    while queue:  # Continue until all reachable nodes are processed
        curr_distance, curr_node = heapq.heappop(queue)  # Get node with smallest known distance

        if curr_distance > distances[curr_node]:  # Skip if we’ve already found a better path
            continue  # Skip this outdated path

        for neighbor, weight in graph[curr_node]:  # Visit all neighboring nodes and edge weights
            distance = curr_distance + weight  # Compute new distance through current node
            if distance < distances[neighbor]:  # Check if this path is shorter than previously known
                distances[neighbor] = distance  # Update with the new shorter distance
                heapq.heappush(queue, (distance, neighbor))  # Enqueue neighbor with updated distance
    return distances  # Return all shortest path distances from the start node

import asyncio
import socket
import random

# Constants for accessing the correct indices in the address info tuple
AI_ADDR_INDEX = 4
ADDR_HOST_INDEX = 0

# A dictionary subclass to manage asynchronous hostname resolution
class Hostnames(dict):
    def __init__(self):
        super().__init__()
        self.tasks = {}      # Tracks ongoing async resolution tasks
        self.started = {}    # Records start time of each resolution
        self.times = {}      # Records resolution duration per hostname

    # Asynchronously resolve a hostname to an IP address
    async def get_hostname(self, hostname):
        loop = asyncio.get_running_loop()

        # Inner coroutine to perform DNS resolution
        async def resolve():
            self.started[hostname] = loop.time()  # Record start time
            try:
                # Perform asynchronous DNS resolution (IPv4 only)
                addrinfos = await loop.getaddrinfo(hostname, None, family=socket.AF_INET)
            except (socket.gaierror, asyncio.TimeoutError):  # Catch errors during resolution
                ip = None
            else:
                # Extract the resolved IP address
                ip = addrinfos[0][AI_ADDR_INDEX][ADDR_HOST_INDEX]
            # Record the total time taken for the resolution
            self.times[hostname] = loop.time() - self.started[hostname]
            return ip

        # If already resolved, return cached result
        if hostname in self:
            return self[hostname]
        else:
            # Reuse an existing task if it's still running
            if hostname in self.tasks:
                task = self.tasks[hostname]
            else:
                # Start a new resolution task
                task = loop.create_task(resolve())
                self.tasks[hostname] = task

            # Ensure task consistency
            assert self.tasks[hostname] is task, f"Task for {hostname} not found"
            assert isinstance(task, asyncio.Task), f"Expected Task, got {type(task)} for {hostname}"

            try:
                # Await the task result
                ip = await task
            except asyncio.CancelledError:
                # If task was cancelled, clean up
                if hostname in self.tasks:
                    del self.tasks[hostname]
                return None
            else:
                # Store the result in the main dictionary
                assert isinstance(ip, str) or ip is None, f"Expected IP address, got {ip} for {hostname}"
                self[hostname] = ip
                return ip

    # Kick off hostname resolution in the background
    def start_get_hostname(self, hostname):
        loop = asyncio.get_running_loop()
        return loop.create_task(self.get_hostname(hostname))

    # Override dictionary access to auto-start resolution if missing
    def __getitem__(self, key):
        if super().__contains__(key):
            return super().__getitem__(key)
        else:
            self.start_get_hostname(key)  # Start resolving in the background
            return None  # Placeholder until resolution completes

# Main coroutine to demonstrate usage
async def main():
    hostnames = Hostnames()

    # Helper function to generate random domain names
    def generate_random_hostname():
        return ''.join(random.choices('abcdefghijklmnopqrstuvwxyz', k=4)) + ".com"

    # Generate a list of random hostnames
    names = [generate_random_hostname() for _ in range(10)]

    # Trigger background resolution for each
    for name in names:
        hostnames[name]

    # Confirm that "example.com" was not pre-resolved
    assert hostnames["example.com"] is None, "Should not have resolved yet"

    # Allow time for DNS resolutions to complete
    await asyncio.sleep(0.2)

    # Print results and resolution times
    for name in names:
        ip = hostnames[name]
        time = hostnames.times.get(name, float('inf'))
        print(f"{name} -> {ip}, time: {time:.3f}s")

# Entry point for script execution
if __name__ == "__main__":
    asyncio.run(main())

# Import necessary libraries
import asyncio  # For asynchronous programming (coroutines, event loop)
import socket   # For network operations, specifically DNS resolution (getaddrinfo)
import random   # For generating random hostnames in the example

# --- Constants for extracting IP address from getaddrinfo results ---
# `getaddrinfo` returns a list of tuples: (family, type, proto, canonname, sockaddr)
# `sockaddr` for AF_INET is a tuple: (address, port)
# So, result[0][4][0] gives the IP address from the first result.
AI_ADDR_INDEX = 4  # Index of the 'sockaddr' tuple in the getaddrinfo result tuple
ADDR_HOST_INDEX = 0 # Index of the host address (IP) within the 'sockaddr' tuple

# --- Hostnames Class ---
# A dictionary-like class to manage asynchronous hostname lookups and cache results.
class Hostnames(dict):
    """
    A dictionary subclass that asynchronously resolves hostnames to IP addresses
    and caches the results. Accessing a hostname via __getitem__ initiates
    the lookup if not already cached or in progress.
    """
    def __init__(self):
        """
        Initializes the Hostnames object.
        - Calls the parent dict initializer.
        - Initializes dictionaries to track ongoing lookup tasks, their start times,
          and the time taken for resolution.
        """
        super().__init__()
        self.tasks = {}   # Stores ongoing asyncio tasks for hostname lookups {hostname: Task}
        self.started = {} # Stores the time when a lookup started {hostname: start_time}
        self.times = {}   # Stores the duration of the lookup {hostname: duration}

    async def get_hostname(self, hostname):
        """
        Asynchronously resolves a given hostname to an IPv4 address.

        If the hostname is already resolved and cached, returns the cached IP.
        If a lookup for this hostname is already in progress, awaits that task.
        Otherwise, starts a new lookup task.

        Args:
            hostname (str): The hostname to resolve.

        Returns:
            str or None: The resolved IPv4 address as a string, or None if
                         resolution fails or is cancelled.
        """
        # Get the currently running asyncio event loop
        loop = asyncio.get_running_loop()

        # --- Inner function to perform the actual DNS resolution ---
        async def resolve():
            """Performs the DNS lookup, records timing, and handles errors."""
            # Record the start time for this specific resolution attempt
            self.started[hostname] = loop.time()
            ip = None # Initialize IP address to None
            try:
                # Perform the asynchronous DNS lookup for IPv4 addresses
                # getaddrinfo can block, so using loop.getaddrinfo makes it async
                addrinfos = await loop.getaddrinfo(hostname, None, family=socket.AF_INET)
                # If successful, extract the IP address from the first result
                # addrinfos is a list of tuples: (family, type, proto, canonname, sockaddr)
                # sockaddr for AF_INET is a tuple: (ip_address, port)
                if addrinfos: # Check if results were returned
                     ip = addrinfos[0][AI_ADDR_INDEX][ADDR_HOST_INDEX]
            except (socket.gaierror, asyncio.TimeoutError) as e:
                # Handle DNS lookup errors (e.g., name not found) or timeouts
                # print(f"Error resolving {hostname}: {e}") # Optional: for debugging
                ip = None # Ensure ip is None on failure
            except Exception as e:
                # Catch any other unexpected errors during resolution
                # print(f"Unexpected error resolving {hostname}: {e}") # Optional: for debugging
                ip = None # Ensure ip is None on failure
            finally:
                # Calculate and store the time taken for the resolution attempt
                if hostname in self.started: # Ensure start time was recorded
                    self.times[hostname] = loop.time() - self.started[hostname]
                # Clean up the start time entry
                # del self.started[hostname] # Optional: Can remove if start time is no longer needed after calculation
            # Return the resolved IP address (or None if resolution failed)
            return ip

        # --- Logic to manage tasks and caching ---

        # Check if the hostname is already resolved and cached in the dictionary
        if hostname in self:
            return self[hostname] # Return the cached IP directly
        else:
            # Hostname not in cache, check if a lookup task is already running
            if hostname in self.tasks:
                # A task for this hostname already exists, get it
                task = self.tasks[hostname]
            else:
                # No existing task, create a new one to resolve the hostname
                # print(f"Creating task for {hostname}") # Optional: for debugging
                task = loop.create_task(resolve())
                # Store the newly created task in the tasks dictionary
                self.tasks[hostname] = task

            # --- Assertions for internal consistency (debugging) ---
            # These checks help ensure the task management logic is working correctly.
            # assert self.tasks[hostname] is task, f"Task for {hostname} not found in dict after adding/retrieving"
            # assert isinstance(task, asyncio.Task), f"Expected Task, got {type(task)} for {hostname}"

            # --- Await the task completion and handle results ---
            try:
                # Wait for the resolution task (either existing or new) to complete
                ip = await task
            except asyncio.CancelledError:
                # If the task gets cancelled externally
                # print(f"Task for {hostname} cancelled.") # Optional: for debugging
                # Remove the cancelled task from the tracking dictionary if it's still there
                if hostname in self.tasks:
                    del self.tasks[hostname]
                return None # Return None as resolution was cancelled
            except Exception as e:
                # Catch any other unexpected error during task execution
                # print(f"Error awaiting task for {hostname}: {e}") # Optional: for debugging
                ip = None # Ensure IP is None if awaiting failed unexpectedly
            finally:
                 # Once the task is finished (successfully, failed, or cancelled),
                 # remove it from the active tasks dictionary.
                 if hostname in self.tasks:
                     del self.tasks[hostname]
                 # Remove start time if resolution failed early or task was cancelled before completion
                 # if hostname in self.started and hostname not in self.times:
                 #    del self.started[hostname] # Optional cleanup

            # --- Post-resolution processing ---
            # Another assertion for type checking (debugging)
            # assert isinstance(ip, str) or ip is None, f"Expected IP address (str) or None, got {ip} for {hostname}"

            # Cache the result (IP address or None) in the dictionary itself
            self[hostname] = ip
            # Return the resolved IP (or None)
            return ip

    def start_get_hostname(self, hostname):
        """
        Creates and starts the asynchronous `get_hostname` task for a given
        hostname but does not wait for it to complete.

        This is useful for initiating multiple lookups concurrently.

        Args:
            hostname (str): The hostname to start resolving.

        Returns:
            asyncio.Task: The created task object for the hostname lookup.
        """
        loop = asyncio.get_running_loop()
        # Create the task using get_hostname coroutine and return it immediately
        return loop.create_task(self.get_hostname(hostname))

    def __getitem__(self, key):
        """
        Overrides dictionary item access (`hostnames[key]`).

        If the hostname (`key`) is already cached, returns the IP address.
        If the hostname is not cached, it *starts* the asynchronous lookup
        using `start_get_hostname` and returns `None` immediately. The actual
        resolution happens in the background. Subsequent calls to __getitem__
        for the same key will return the cached IP once resolution is complete.

        Args:
            key (str): The hostname to look up.

        Returns:
            str or None: The cached IP address if available, otherwise None.
        """
        # Check if the key (hostname) is already in the dictionary (cache)
        if super().__contains__(key):
            # If yes, return the cached value (IP or None if resolution failed previously)
            return super().__getitem__(key)
        else:
            # If not in cache, initiate the background resolution process
            # Note: This call starts the task but doesn't wait for it.
            self.start_get_hostname(key)
            # Return None immediately, indicating the result is not yet available.
            # The actual result will be available in the dictionary later
            # once the background task completes.
            return None

# --- Example Usage ---
async def main():
    """
    Demonstrates the usage of the Hostnames class.
    """
    # Create an instance of the Hostnames cache
    hostnames = Hostnames()

    # Helper function to generate random (likely non-existent) hostnames for testing
    def generate_random_hostname():
        """Generates a random 4-letter .com hostname."""
        return ''.join(random.choices('abcdefghijklmnopqrstuvwxyz', k=4)) + ".com"

    # Generate a list of 10 random hostnames
    names = [generate_random_hostname() for _ in range(10)]
    # Add a known valid hostname for comparison
    names.append("example.com")
    names.append("google.com") # Add another valid one

    print("Starting hostname resolution tasks...")
    # Iterate through the list of hostnames
    for name in names:
        # Accessing the hostname via hostnames[name] triggers __getitem__
        # This either returns a cached value (unlikely on the first pass)
        # or starts the background resolution task and returns None.
        _ = hostnames[name] # We discard the initial None return value

    # Assertion: Check that a known hostname hasn't resolved *instantly*.
    # Accessing it starts the task, but it shouldn't be complete yet.
    # assert hostnames["example.com"] is None, "Should not have resolved 'example.com' instantaneously"
    print(f"Initial access for example.com: {hostnames['example.com']} (expected None)")


    # Wait for a short period to allow background DNS lookups to proceed/complete.
    # The required time depends on network latency and DNS server responsiveness.
    print("\nWaiting for resolutions to complete...")
    await asyncio.sleep(1.0) # Increased sleep time for potentially slower lookups

    print("\nChecking results:")
    # Iterate through the names again to retrieve the results
    for name in names:
        # Access the hostname again. This time, __getitem__ should return
        # the cached result (IP address or None) if the task has finished.
        ip = hostnames[name]
        # Get the resolution time from the `times` dictionary, default to infinity if not found
        time_taken = hostnames.times.get(name, float('inf'))
        # Format the time for display
        time_str = f"{time_taken:.3f}s" if time_taken != float('inf') else "N/A (pending or failed early)"
        # Print the hostname, its resolved IP (or None), and the time taken
        print(f"{name:<20} -> {ip:<15} (took {time_str})")

# --- Script Entry Point ---
if __name__ == "__main__":
    # Run the main asynchronous function using asyncio.run()
    # This starts the asyncio event loop.
    asyncio.run(main())

import asyncio
import socket
import pytest
from unittest.mock import patch, AsyncMock

from your_module import Hostnames  # Replace 'your_module' with the actual filename

@pytest.mark.asyncio
async def test_get_hostname_success():
    hostnames = Hostnames()

    fake_ip = "93.184.216.34"
    fake_hostname = "example.com"
    addrinfo_mock = [(None, None, None, None, (fake_ip, 0))]

    with patch("asyncio.get_running_loop") as mock_loop:
        loop = asyncio.get_event_loop()
        mock_loop.return_value = loop

        with patch.object(loop, "getaddrinfo", AsyncMock(return_value=addrinfo_mock)):
            ip = await hostnames.get_hostname(fake_hostname)

    assert ip == fake_ip
    assert hostnames[fake_hostname] == fake_ip
    assert fake_hostname in hostnames.times
    assert hostnames.times[fake_hostname] >= 0

@pytest.mark.asyncio
async def test_get_hostname_cached():
    hostnames = Hostnames()
    hostnames["example.com"] = "1.2.3.4"

    ip = await hostnames.get_hostname("example.com")
    assert ip == "1.2.3.4"

@pytest.mark.asyncio
async def test_get_hostname_error():
    hostnames = Hostnames()
    hostname = "nonexistent.domain"

    with patch("asyncio.get_running_loop") as mock_loop:
        loop = asyncio.get_event_loop()
        mock_loop.return_value = loop

        with patch.object(loop, "getaddrinfo", AsyncMock(side_effect=socket.gaierror)):
            ip = await hostnames.get_hostname(hostname)

    assert ip is None
    assert hostname in hostnames.times

@pytest.mark.asyncio
async def test_get_hostname_concurrent():
    hostnames = Hostnames()
    hostname = "example.com"
    fake_ip = "8.8.8.8"
    addrinfo_mock = [(None, None, None, None, (fake_ip, 0))]

    with patch("asyncio.get_running_loop") as mock_loop:
        loop = asyncio.get_event_loop()
        mock_loop.return_value = loop

        with patch.object(loop, "getaddrinfo", AsyncMock(return_value=addrinfo_mock)):
            task1 = asyncio.create_task(hostnames.get_hostname(hostname))
            task2 = asyncio.create_task(hostnames.get_hostname(hostname))
            ip1, ip2 = await asyncio.gather(task1, task2)

    assert ip1 == ip2 == fake_ip
    assert isinstance(hostnames.tasks[hostname], asyncio.Task)

@pytest.mark.asyncio
async def test_cancelled_resolution():
    hostnames = Hostnames()
    hostname = "cancelled.com"

    with patch("asyncio.get_running_loop") as mock_loop:
        loop = asyncio.get_event_loop()
        mock_loop.return_value = loop

        async def delayed_resolve():
            await asyncio.sleep(1)
            return [(None, None, None, None, ("1.1.1.1", 0))]

        with patch.object(loop, "getaddrinfo", delayed_resolve):
            task = asyncio.create_task(hostnames.get_hostname(hostname))
            await asyncio.sleep(0.1)
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

            assert hostname not in hostnames

@pytest.mark.asyncio
async def test_getitem_triggers_resolution():
    hostnames = Hostnames()
    hostname = "lazy.com"

    with patch.object(hostnames, "start_get_hostname", return_value=asyncio.Future()) as mock_start:
        value = hostnames[hostname]

    assert value is None
    mock_start.assert_called_once_with(hostname)

pip install pytest pytest-asyncio

import asyncio
import socket
import random
import time
import unittest
from unittest.mock import patch, AsyncMock, MagicMock

# The code to be tested (ensure it's accessible, e.g., in the same file or imported)
AI_ADDR_INDEX = 4
ADDR_HOST_INDEX = 0

class Hostnames(dict):
    def __init__(self):
        super().__init__()
        self.tasks = {}
        self.started = {}
        self.times = {}

    async def get_hostname(self, hostname):
        loop = asyncio.get_running_loop()
        async def resolve():
            self.started[hostname] = loop.time()
            try:
                # Give other tasks a chance to run, simulating real I/O wait
                await asyncio.sleep(0.01)
                addrinfos = await loop.getaddrinfo(hostname, None, family=socket.AF_INET)
            except (socket.gaierror, asyncio.TimeoutError):
                ip = None
            else:
                if addrinfos: # Ensure addrinfos is not empty
                    ip = addrinfos[0][AI_ADDR_INDEX][ADDR_HOST_INDEX]
                else:
                    ip = None # Handle case where getaddrinfo returns empty list
            self.times[hostname] = loop.time() - self.started.get(hostname, self.started.get(hostname, loop.time())) # ensure started exists
            return ip

        if hostname in self:
            return self[hostname]
        else:
            if hostname in self.tasks:
                task = self.tasks[hostname]
            else:
                task = loop.create_task(resolve())
                self.tasks[hostname] = task

            assert self.tasks[hostname] is task, f"Task for {hostname} not found"
            assert isinstance(task, asyncio.Task), f"Expected Task, got {type(task)} for {hostname}"

            try:
                ip = await task
            except asyncio.CancelledError:
                if hostname in self.tasks: # Task might already be removed if cancellation is very fast
                    del self.tasks[hostname]
                # Do not store None on cancellation, as it's not a resolution result
                # self[hostname] = None # Original code stores None, let's keep for now but it's debatable
                return None # Explicitly return None on cancellation
            else:
                assert isinstance(ip, str) or ip is None, f"Expected IP address, got {ip} for {hostname}"
                self[hostname] = ip
                # Task is done, remove it
                if hostname in self.tasks:
                    del self.tasks[hostname]
                return ip

    def start_get_hostname(self, hostname):
        loop = asyncio.get_running_loop()
        # Ensure that calling start_get_hostname doesn't re-create a task if one is pending
        # or if the hostname is already resolved. get_hostname handles this.
        return loop.create_task(self.get_hostname(hostname))

    def __getitem__(self, key):
        if super().__contains__(key):
            return super().__getitem__(key)
        else:
            # Only start a new task if one isn't already running for this key
            if key not in self.tasks:
                 self.start_get_hostname(key) # This will call get_hostname, which handles task creation
            return None # Always return None immediately if not resolved yet

    def __setitem__(self, key, value):
        super().__setitem__(key, value)
        # If we are setting a value, it implies resolution is complete or overridden.
        # Clean up any pending tasks for this hostname.
        if key in self.tasks:
            task = self.tasks.pop(key)
            if not task.done():
                task.cancel()
        if key in self.started:
            del self.started[key]
        if key in self.times:
            del self.times[key]


class TestHostnames(unittest.IsolatedAsyncioTestCase):

    async def asyncSetUp(self):
        self.hostnames = Hostnames()
        # It's good practice to have a fresh loop for each test,
        # IsolatedAsyncioTestCase does this.
        # self.loop = asyncio.get_running_loop() # No need, IsolatedAsyncioTestCase provides loop

    def _get_mock_addrinfo(self, ip_address="192.168.1.1"):
        # Structure: (family, type, proto, canonname, sockaddr)
        # sockaddr for AF_INET is (host, port)
        # The code accesses addrinfos[0][AI_ADDR_INDEX][ADDR_HOST_INDEX]
        # So, addrinfos[0][4][0]
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', ('', 0, 0, 0, (ip_address, 0)))]


    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_get_single_hostname_success(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = self._get_mock_addrinfo("1.2.3.4")

        hostname = "test.com"
        ip = await self.hostnames.get_hostname(hostname)

        self.assertEqual(ip, "1.2.3.4")
        self.assertEqual(self.hostnames[hostname], "1.2.3.4")
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)
        self.assertIn(hostname, self.hostnames.times)
        self.assertIn(hostname, self.hostnames.started)
        self.assertNotIn(hostname, self.hostnames.tasks) # Task should be removed after completion

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_get_hostname_caching(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = self._get_mock_addrinfo("5.6.7.8")
        hostname = "cache.com"

        ip1 = await self.hostnames.get_hostname(hostname)
        self.assertEqual(ip1, "5.6.7.8")
        self.assertEqual(self.hostnames[hostname], "5.6.7.8")
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)

        # Call again, should use cache
        ip2 = await self.hostnames.get_hostname(hostname)
        self.assertEqual(ip2, "5.6.7.8")
        mock_getaddrinfo.assert_called_once() # Should still be called only once

        self.assertNotIn(hostname, self.hostnames.tasks)

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_get_hostname_failure_gaierror(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.side_effect = socket.gaierror("Test gai error")
        hostname = "fail.com"

        ip = await self.hostnames.get_hostname(hostname)
        self.assertIsNone(ip)
        self.assertIsNone(self.hostnames[hostname]) # Now stores None on failure
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)
        self.assertIn(hostname, self.hostnames.times)
        self.assertNotIn(hostname, self.hostnames.tasks)

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_get_hostname_failure_timeout(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.side_effect = asyncio.TimeoutError("Test timeout error")
        hostname = "timeout.com"

        ip = await self.hostnames.get_hostname(hostname)
        self.assertIsNone(ip)
        self.assertIsNone(self.hostnames[hostname])
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)
        self.assertIn(hostname, self.hostnames.times)
        self.assertNotIn(hostname, self.hostnames.tasks)

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_get_hostname_empty_addrinfo(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = [] # Simulate getaddrinfo returning an empty list
        hostname = "empty.com"

        ip = await self.hostnames.get_hostname(hostname)
        self.assertIsNone(ip)
        self.assertIsNone(self.hostnames[hostname])
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)
        self.assertIn(hostname, self.hostnames.times)
        self.assertNotIn(hostname, self.hostnames.tasks)


    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_concurrent_get_hostname_same_host(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = self._get_mock_addrinfo("10.0.0.1")
        hostname = "concurrent.com"

        # Make resolve take a bit longer to ensure tasks are created before first one finishes
        original_resolve_delay = 0.01 # from Hostnames class
        # Temporarily increase delay inside resolve for this test if needed,
        # or ensure mock_getaddrinfo itself has a delay.
        # For this, we'll rely on the small asyncio.sleep(0.01) in resolve.

        task1 = asyncio.create_task(self.hostnames.get_hostname(hostname))
        task2 = asyncio.create_task(self.hostnames.get_hostname(hostname))

        ip1 = await task1
        ip2 = await task2

        self.assertEqual(ip1, "10.0.0.1")
        self.assertEqual(ip2, "10.0.0.1")
        self.assertEqual(self.hostnames[hostname], "10.0.0.1")
        # getaddrinfo should only be called once because the second call should find the task
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)
        self.assertNotIn(hostname, self.hostnames.tasks)


    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_getitem_initiates_resolution(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = self._get_mock_addrinfo("11.0.0.1")
        hostname = "getitem.com"

        # First __getitem__ call
        ip_initial = self.hostnames[hostname]
        self.assertIsNone(ip_initial) # Should return None immediately
        self.assertIn(hostname, self.hostnames.tasks) # Task should be created

        # Wait for the task to complete
        resolved_ip = await self.hostnames.tasks[hostname] # Await the task directly for testing

        self.assertEqual(resolved_ip, "11.0.0.1")
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)

        # Second __getitem__ call after resolution
        ip_after_resolution = self.hostnames[hostname]
        self.assertEqual(ip_after_resolution, "11.0.0.1")
        self.assertNotIn(hostname, self.hostnames.tasks) # Task should be removed

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_start_get_hostname(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = self._get_mock_addrinfo("12.0.0.1")
        hostname = "start.com"

        task = self.hostnames.start_get_hostname(hostname)
        self.assertIsInstance(task, asyncio.Task)

        ip = await task
        self.assertEqual(ip, "12.0.0.1")
        self.assertEqual(self.hostnames[hostname], "12.0.0.1")
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)
        self.assertNotIn(hostname, self.hostnames.tasks)

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_task_cancellation(self, mock_getaddrinfo: AsyncMock):
        hostname = "cancel.com"

        # Make getaddrinfo slow to allow cancellation
        async def slow_getaddrinfo(*args, **kwargs):
            await asyncio.sleep(0.2) # Make it slow enough to be cancelled
            return self._get_mock_addrinfo("1.1.1.1")
        mock_getaddrinfo.side_effect = slow_getaddrinfo

        task = asyncio.create_task(self.hostnames.get_hostname(hostname))
        self.assertIn(hostname, self.hostnames.tasks)

        await asyncio.sleep(0.01) # Give the task a moment to start
        task.cancel()

        with self.assertRaises(asyncio.CancelledError):
             await task # This should re-raise CancelledError if caught and re-raised by task logic
                        # or just complete if cancellation is handled internally by get_hostname

        # Check the result of get_hostname after cancellation
        # The `get_hostname` method itself catches CancelledError and should return None.
        # To test this properly, we'd call get_hostname again or check its internal state handling.
        # Let's try calling it again, it should NOT try to resolve if the task was cancelled and removed.
        # However, the current implementation of get_hostname returns None directly if await task raises CancelledError.

        # Re-create the scenario slightly differently:
        self.hostnames = Hostnames() # Reset state for clarity
        task_to_cancel = self.hostnames.start_get_hostname(hostname)

        await asyncio.sleep(0.02) # allow task to start the resolve() coroutine

        self.assertTrue(hostname in self.hostnames.tasks)
        internal_task = self.hostnames.tasks[hostname]
        internal_task.cancel() # Cancel the inner task created by resolve()

        result = await task_to_cancel # This is the outer task from start_get_hostname

        self.assertIsNone(result) # get_hostname should return None on cancellation
        self.assertNotIn(hostname, self.hostnames) # Should not store None in self on cancellation
        self.assertNotIn(hostname, self.hostnames.tasks) # Task should be removed
        mock_getaddrinfo.assert_called_once() # It should have been called once before cancellation hit

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_timing_information_success(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = self._get_mock_addrinfo("13.0.0.1")
        hostname = "time.com"

        start_time = asyncio.get_running_loop().time()
        await self.hostnames.get_hostname(hostname)
        end_time = asyncio.get_running_loop().time()

        self.assertIn(hostname, self.hostnames.started)
        self.assertIn(hostname, self.hostnames.times)
        self.assertGreaterEqual(self.hostnames.started[hostname], start_time - 0.001) # allow small diff
        self.assertLessEqual(self.hostnames.started[hostname], end_time)

        duration = self.hostnames.times[hostname]
        self.assertGreaterEqual(duration, 0)
        # The internal sleep is 0.01
        self.assertGreaterEqual(duration, 0.009) # Should be at least the sleep duration
        self.assertLessEqual(duration, end_time - self.hostnames.started[hostname] + 0.001)

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_timing_information_failure(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.side_effect = socket.gaierror("Timing fail")
        hostname = "timefail.com"

        start_time = asyncio.get_running_loop().time()
        await self.hostnames.get_hostname(hostname)
        end_time = asyncio.get_running_loop().time()

        self.assertIn(hostname, self.hostnames.started)
        self.assertIn(hostname, self.hostnames.times)
        self.assertGreaterEqual(self.hostnames.started[hostname], start_time - 0.001)
        self.assertLessEqual(self.hostnames.started[hostname], end_time)

        duration = self.hostnames.times[hostname]
        self.assertGreaterEqual(duration, 0)
        self.assertGreaterEqual(duration, 0.009) # Should be at least the sleep duration

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_getitem_multiple_calls_before_resolution(self, mock_getaddrinfo: AsyncMock):
        mock_getaddrinfo.return_value = self._get_mock_addrinfo("14.0.0.1")
        hostname = "multiitem.com"

        # __getitem__ calls should return None and only trigger one actual lookup
        self.assertIsNone(self.hostnames[hostname]) # First call, starts task
        self.assertIn(hostname, self.hostnames.tasks)
        task_id_1 = id(self.hostnames.tasks[hostname])

        self.assertIsNone(self.hostnames[hostname]) # Second call, should see existing task
        self.assertIn(hostname, self.hostnames.tasks)
        task_id_2 = id(self.hostnames.tasks[hostname])
        self.assertEqual(task_id_1, task_id_2) # Same task instance

        # Wait for resolution
        ip = await self.hostnames.tasks[hostname]
        self.assertEqual(ip, "14.0.0.1")

        # Now __getitem__ should return the IP
        self.assertEqual(self.hostnames[hostname], "14.0.0.1")
        mock_getaddrinfo.assert_called_once_with(hostname, None, family=socket.AF_INET)
        self.assertNotIn(hostname, self.hostnames.tasks)

    @patch('asyncio.DefaultEventLoopPolicy._event_loop_policy._local._loop.getaddrinfo')
    async def test_setitem_clears_pending_task(self, mock_getaddrinfo: AsyncMock):
        hostname = "setitem.com"

        # Make getaddrinfo very slow
        resolve_event = asyncio.Event()
        async def very_slow_getaddrinfo(*args, **kwargs):
            await resolve_event.wait() # Will hang until event is set
            return self._get_mock_addrinfo("1.1.1.1")
        mock_getaddrinfo.side_effect = very_slow_getaddrinfo

        # Start resolution via __getitem__
        self.assertIsNone(self.hostnames[hostname])
        self.assertIn(hostname, self.hostnames.tasks)
        original_task = self.hostnames.tasks[hostname]
        self.assertFalse(original_task.done())

        # Now, directly set the item
        self.hostnames[hostname] = "127.0.0.1"
        self.assertEqual(self.hostnames[hostname], "127.0.0.1")

        # The original task should have been cancelled and removed
        self.assertNotIn(hostname, self.hostnames.tasks)
        self.assertTrue(original_task.cancelled() or original_task.done()) # It might finish quickly if cancel is called late

        # Allow the mocked getaddrinfo to proceed to ensure no errors if it was too late to cancel
        resolve_event.set()
        await asyncio.sleep(0.01) # give it a chance to finish if it wasn't cancelled in time

        # getaddrinfo should not have been called if task was effectively cancelled before await
        # or it might have been called if __setitem__ is slow. The key is that the task is gone.
        # The critical check is that self.hostnames.tasks is empty.

        # Re-check the value, should still be the one we set
        self.assertEqual(self.hostnames[hostname], "127.0.0.1")


async def main_for_manual_testing():
    # This is for running the original main logic if needed, not part of unittest
    hostnames = Hostnames()
    def generate_random_hostname():
        return ''.join(random.choices('abcdefghijklmnopqrstuvwxyz', k=4)) + ".com"

    # Use some known good and bad hostnames for more predictable manual testing
    names = ["google.com", "nonexistentfakedomain123abc.com", "github.com"]
    # names.extend([generate_random_hostname() for _ in range(3)]) # Add some random ones

    print("Initial __getitem__ calls (should be None and start tasks):")
    for name in names:
        print(f"{name} -> {hostnames[name]}") # This will print None and queue tasks

    print("\nWaiting for resolutions...")
    # Wait for tasks to complete, check periodically.
    # A more robust way would be to gather the tasks if __getitem__ returned them.
    # Since it doesn't, we rely on an arbitrary sleep or check tasks dict.

    # Let's try to await the tasks if they exist
    # This part is a bit tricky because __getitem__ doesn't return the task
    # We could iterate self.hostnames.tasks, but that's internal.
    # For this manual test, let's just sleep and then print.

    # A better approach for testing this part of main would be to call get_hostname
    # and gather results.

    # Create tasks explicitly for waiting
    tasks = []
    for name in names:
        # If already resolved by previous __getitem__ and sleep, this will be fast
        # If not, it will await the existing task or create a new one.
        tasks.append(asyncio.create_task(hostnames.get_hostname(name)))

    await asyncio.gather(*tasks, return_exceptions=True) # Wait for all to complete

    print("\nResolved IPs:")
    assert hostnames["example.com"] is None, "Should not have resolved yet unless explicitly added"
    # The original main had this assertion, but example.com was never added to `names`.
    # Let's add it to make the assertion meaningful or remove the assertion.
    # For now, I'll comment it out as it wasn't in the initial names list for this adapted test.
    # hostnames["example.com"] # This would trigger its resolution if we want to test it.
    # await hostnames.get_hostname("example.com")

    await asyncio.sleep(0.2) # Additional sleep from original main

    for name in names:
        ip = hostnames[name] # Now this should return the resolved IP or None
        time_taken = hostnames.times.get(name, float('inf'))
        print(f"{name} -> {ip}, time: {time_taken:.3f}s")

    # Test example.com as in original main
    print("\nTesting example.com (as per original main logic):")
    hostnames["example.com"] # Trigger resolution
    assert hostnames["example.com"] is None, "example.com should not have resolved yet (first __getitem__)"
    await asyncio.sleep(0.2) # Give it time
    ip_example = await hostnames.get_hostname("example.com") # Ensure it resolves
    time_example = hostnames.times.get("example.com", float('inf'))
    print(f"example.com -> {ip_example}, time: {time_example:.3f}s")
    # The assertion `hostnames["example.com"] is None` in the original main after the loop
    # that calls `hostnames[name]` is racy. It might be resolved or not.
    # The second loop `for name in names: ip = hostnames[name]` is where it expects resolution.

if __name__ == '__main__':
    # To run the unittests:
    unittest.main(argv=['first-arg-is-ignored'], exit=False)

    # To run the manual testing part (similar to original main):
    # print("\n--- Running Manual Main Test ---")
    # asyncio.run(main_for_manual_testing())