def test_socket_can_shake_hands():
    mk(('echo.sock', ''))
    socket = make_socket()
    response = socket.shake_hands()
    expected = '15:10:xhr-polling'
    actual = response.body.split(':', 1)[1]
    assert actual == expected, actual

def test_can_put_onto_buffer():
    mk(('echo.sock.spt', 'socket.send(socket.recv())'))
    buffer = Buffer(make_socket(), 'foo')
    expected = [FFFD+'4'+FFFD+'1:::']
    buffer.put(Message.from_bytes('1:::'))
    actual = list(buffer.flush())
    assert actual == expected, actual

def test_root_defaults_to_cwd():
    mk()
    c = Configurable()
    c.configure([])
    expected = os.getcwd()
    actual = c.root
    assert actual == expected, actual

def test_redirect_has_only_location():
    mk(('index.html.spt', "from aspen import Response\n[---]\nrequest.redirect('http://elsewhere', code=304)\n[---]\n"))
    actual = handle()
    assert actual.code == 304
    headers = actual.headers
    assert len(headers) == 1, headers
    assert headers.get('Location') is not None, headers

def test_unavailable_knob_works():
    mk( '.aspen'
      , ('.aspen/foo.py', 'bar = "baz"')
      , ('index.html', "import fooGreetings, {{ foo.bar }}!")
       )
    actual = handle('/', '--unavailable=1').code
    assert actual == 503, actual

def test_configurable_sees_root_option():
    mk()
    c = Configurable()
    c.configure(['--root', FSFIX])
    expected = os.getcwd()
    actual = c.root
    assert actual == expected, actual

def test_www_root_defaults_to_cwd():
    mk()
    c = Configurable()
    c.configure([])
    expected = os.path.realpath(os.getcwd())
    actual = c.www_root
    assert actual == expected, actual

def test_can_put_onto_buffer():
    mk(("echo.sock", "socket.send(socket.recv())"))
    buffer = Buffer(make_socket(), "foo")
    expected = [FFFD + "4" + FFFD + "1:::"]
    buffer.put(Message.from_bytes("1:::"))
    actual = list(buffer.flush())
    assert actual == expected, actual

def test_resources_can_import_from_dot_aspen():
    mk( '.aspen'
      , ('.aspen/foo.py', 'bar = "baz"')
      , ('index.html', "import fooGreetings, {{ foo.bar }}!")
       )
    expected = "Greetings, baz!"
    actual = handle('/', '--project_root=.aspen').body
    assert actual == expected, actual

def make_transport(content='', state=0):
    mk(('echo.sock', content))
    socket = make_socket()
    transport = XHRPollingTransport(socket)
    transport.timeout = 0.05 # for testing, could screw up the test
    if state == 1:
        transport.respond(Request(uri='/echo.sock'))
    return transport

def test_channel_can_have_sockets_added_to_it():
    mk(('echo.sock', 'channel.send(channel.recv())'))
    socket = make_socket()
    channel = Channel('foo', ThreadedBuffer)
    channel.add(socket)

    expected = [socket]
    actual = list(channel)
    assert actual == expected, actual

def test_two_sockets_are_instantiable():
    mk(('echo.sock', ''))

    socket1 = make_socket()
    socket2 = make_socket()

    expected = (Socket, Socket)
    actual = (socket1.__class__, socket2.__class__)
    assert actual == expected, actual

def test_unavailable_knob_sets_retry_after_on_website():
    mk( '.aspen'
      , ('.aspen/foo.py', 'bar = "baz"')
      , ('index.html', "import fooGreetings, {{ foo.bar }}!")
       )
    actual = handle('/', '--unavailable=10').headers['Retry-After']
    expected = datetime.datetime.utcnow().strftime('%a, %d %b %Y')
    assert actual.startswith(expected), actual
    assert actual.endswith(' +0000'), actual

def test_path_part_params_are_available():
    mk(('/foo/index.html.spt', """
if 'b' in path.parts[0].params:
    a = path.parts[0].params['a']
[---]
%(a)s"""))
    expected = "3"
    actual = handle('/foo;a=1;b;a=3/').body
    assert actual == expected, actual + " isn't " + expected

def test_resources_can_import_from_dot_aspen():
    mk( '.aspen'
      , ('.aspen/foo.py', 'bar = "baz"')
      , ('index.html.spt', "from foo import bar\n[---]\nGreetings, %(bar)s!")
       )
    expected = "Greetings, baz!"
    project_root = os.path.join(FSFIX, '.aspen')
    actual = handle('/', '--project_root='+project_root).body
    assert actual == expected, actual

def test_double_failure_still_sets_response_dot_request():
    mk( '.aspen'
      , ('.aspen/foo.py', """
def bar(response):
    response.request
""")
      , ('.aspen/hooks.conf', 'foo:bar')
      , ('index.html', "raise heck")
       )

def test_socket_can_barely_function():
    mk(('echo.sock', 'socket.send("Greetings, program!")'))

    socket = make_socket()
    socket.tick()

    expected = FFFD+'33'+FFFD+'3::/echo.sock:Greetings, program!'
    actual = socket._recv().next()
    assert actual == expected, actual

def test_socket_can_echo():
    mk(('echo.sock', 'socket.send(socket.recv())'))

    with SocketInThread() as socket:
        socket._send('3::/echo.sock:Greetings, program!')
        time.sleep(0.05) # give the resource time to tick

        expected = FFFD+'33'+FFFD+'3::/echo.sock:Greetings, program!'
        actual = socket._recv().next()
        assert actual == expected, actual

def test_ConfigurationError_NOT_raised_if_no_cwd_but_do_have___root():
    mk()
    foo = os.getcwd()
    os.chdir(FSFIX)
    os.rmdir(os.getcwd())
    c = Configurable()
    c.configure(['--root', foo])
    expected = foo
    actual = c.root
    assert actual == expected, actual

def test_channel_passes_send_on_to_one_socket():
    mk(('echo.sock', ''))
    socket = make_socket()
    channel = Channel('foo', ThreadedBuffer)
    channel.add(socket)
    channel.send('foo')

    expected = deque([Message.from_bytes('3::/echo.sock:foo')])
    actual = socket.outgoing.queue
    assert actual == expected, actual