/ tests / protocol_test.py
protocol_test.py
  1  from unittest import mock
  2  import mocket
  3  import pytest
  4  import adafruit_requests
  5  
  6  ip = "1.2.3.4"
  7  host = "wifitest.adafruit.com"
  8  path = "/testwifi/index.html"
  9  text = b"This is a test of Adafruit WiFi!\r\nIf you can read this, its working :)"
 10  response = b"HTTP/1.0 200 OK\r\nContent-Length: 70\r\n\r\n" + text
 11  
 12  
 13  def test_get_https_no_ssl():
 14      pool = mocket.MocketPool()
 15      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 16      sock = mocket.Mocket(response)
 17      pool.socket.return_value = sock
 18  
 19      s = adafruit_requests.Session(pool)
 20      with pytest.raises(RuntimeError):
 21          r = s.get("https://" + host + path)
 22  
 23  
 24  def test_get_https_text():
 25      pool = mocket.MocketPool()
 26      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 27      sock = mocket.Mocket(response)
 28      pool.socket.return_value = sock
 29      ssl = mocket.SSLContext()
 30  
 31      s = adafruit_requests.Session(pool, ssl)
 32      r = s.get("https://" + host + path)
 33  
 34      sock.connect.assert_called_once_with((host, 443))
 35  
 36      sock.send.assert_has_calls(
 37          [
 38              mock.call(b"GET"),
 39              mock.call(b" /"),
 40              mock.call(b"testwifi/index.html"),
 41              mock.call(b" HTTP/1.1\r\n"),
 42          ]
 43      )
 44      sock.send.assert_has_calls(
 45          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 46      )
 47      assert r.text == str(text, "utf-8")
 48  
 49      # Close isn't needed but can be called to release the socket early.
 50      r.close()
 51  
 52  
 53  def test_get_http_text():
 54      pool = mocket.MocketPool()
 55      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 56      sock = mocket.Mocket(response)
 57      pool.socket.return_value = sock
 58  
 59      s = adafruit_requests.Session(pool)
 60      r = s.get("http://" + host + path)
 61  
 62      sock.connect.assert_called_once_with((ip, 80))
 63  
 64      sock.send.assert_has_calls(
 65          [
 66              mock.call(b"GET"),
 67              mock.call(b" /"),
 68              mock.call(b"testwifi/index.html"),
 69              mock.call(b" HTTP/1.1\r\n"),
 70          ]
 71      )
 72      sock.send.assert_has_calls(
 73          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
 74      )
 75      assert r.text == str(text, "utf-8")
 76  
 77  
 78  def test_get_close():
 79      """Test that a response can be closed without the contents being accessed."""
 80      pool = mocket.MocketPool()
 81      pool.getaddrinfo.return_value = ((None, None, None, None, (ip, 80)),)
 82      sock = mocket.Mocket(response)
 83      pool.socket.return_value = sock
 84  
 85      s = adafruit_requests.Session(pool)
 86      r = s.get("http://" + host + path)
 87  
 88      sock.connect.assert_called_once_with((ip, 80))
 89  
 90      sock.send.assert_has_calls(
 91          [
 92              mock.call(b"GET"),
 93              mock.call(b" /"),
 94              mock.call(b"testwifi/index.html"),
 95              mock.call(b" HTTP/1.1\r\n"),
 96          ]
 97      )
 98      sock.send.assert_has_calls(
 99          [mock.call(b"Host: "), mock.call(b"wifitest.adafruit.com"),]
100      )
101      r.close()