text: # Copyright (c) 2013-2019 Siphon Contributors.
  # Distributed under the terms of the BSD 3-Clause License.
  # SPDX-License-Identifier: BSD-3-Clause
  """Test Siphon's base HTTP helper functionality."""
  
  from datetime import datetime, timedelta
  
  import pytest
  
  from siphon.http_util import (DataQuery, HTTPEndPoint, HTTPError,
                                parse_iso_date, session_manager, utc)
  import siphon.testing
  
  recorder = siphon.testing.get_recorder(__file__)
  
  
  @recorder.use_cassette('top_thredds_catalog')
  def test_urlopen():
      """Test siphon's urlopen wrapper."""
      fobj = session_manager.urlopen('http://thredds-test.unidata.ucar.edu/thredds/catalog.xml')
      assert fobj.read(2) == b'<?'
  
  
  @recorder.use_cassette('top_thredds_catalog')
  def test_session():
      """Test that http sessions contain the proper user agent."""
      session = session_manager.create_session()
      resp = session.get('http://thredds-test.unidata.ucar.edu/thredds/catalog.xml')
      assert resp.request.headers['user-agent'].startswith('Siphon')
  
  
  @recorder.use_cassette('top_thredds_catalog')
  def test_session_options():
      """Test that http sessions receive proper options."""
      auth = ('foo', 'bar')
      session_manager.set_session_options(auth=auth)
      try:
          session = session_manager.create_session()
          assert session.auth == auth
      finally:
          session_manager.set_session_options()
  
  
  def test_parse_iso():
      """Test parsing ISO-formatted dates."""
      parsed = parse_iso_date('2015-06-15T12:00:00Z')
      assert parsed.utcoffset() == timedelta(0)
      assert parsed == datetime(2015, 6, 15, 12, tzinfo=utc)
  
  
  def test_data_query_basic():
      """Test forming a basic query."""
      dr = DataQuery()
      dr.all_times()
      dr.variables('foo', 'bar')
      query = repr(dr)  # To check repr for once
      assert 'temporal=all' in query
      assert 'var=foo' in query
      assert 'var=bar' in query
  
  
  def test_data_query_repeated_vars():
      """Test a query properly de-duplicates variables."""
      dr = DataQuery()
      dr.variables('foo', 'bar')
      dr.variables('foo')
      query = str(dr)
      assert query.count('foo') == 1
      assert query.count('bar') == 1
  
  
  def test_data_query_time_reset():
      """Test query with multiple time-type query fields."""
      dr = DataQuery().all_times().time(datetime.utcnow())
      query = str(dr)
      assert query.startswith('time='), 'Bad string: ' + query
      assert query.count('=') == 1
  
  
  def test_data_query_time_reset2():
      """Test that time queries replace each other."""
      dr = DataQuery().time(datetime.utcnow()).all_times()
      assert str(dr) == 'temporal=all'
  
  
  def test_data_query_time_reset3():
      """Test that time queries replace each other with time range."""
      dt = datetime(2015, 6, 12, 12, 0, 0)
      dt2 = datetime(2015, 6, 13, 12, 0, 0)
      dr = DataQuery().all_times().time_range(dt, dt2)
      query = str(dr)
      assert 'time_start=2015-06-12T12' in query, 'Bad string: ' + query
      assert 'time_end=2015-06-13T12' in query, 'Bad string: ' + query
      assert query.count('=') == 2
  
  
  def test_data_query_time_format():
      """Test that time queries are properly formatted."""
      dt = datetime(2015, 6, 15, 12, 0, 0)
      dr = DataQuery().time(dt)
      query = str(dr)
      assert query == 'time=2015-06-15T12%3A00%3A00'
  
  
  def test_data_query_bad_time_range():
      """Test that time queries are properly formatted."""
      with pytest.warns(UserWarning):
          dt = datetime(2015, 6, 15, 12, 0, 0)
          dt2 = datetime(2015, 6, 14, 12, 0, 0)
          DataQuery().time_range(dt, dt2)
  
  
  def test_data_query_spatial_reset():
      """Test that spatial queries reset each other."""
      dr = DataQuery().lonlat_box(1, 2, 3, 4).lonlat_point(-1, -2)
      query = str(dr)
      assert 'latitude=-2' in query
      assert 'longitude=-1' in query
      assert query.count('=') == 2
  
  
  def test_data_query_spatial_reset2():
      """Test more that spatial queries reset each other."""
      dr = DataQuery().lonlat_point(-1, -2).lonlat_box(1, 2, 3, 4)
      query = str(dr)
      assert 'south=3' in query
      assert 'north=4' in query
      assert 'west=1' in query
      assert 'east=2' in query
      assert query.count('=') == 4
  
  
  def test_data_query_iter():
      """Test converting a query to a dictionary."""
      dt = datetime.utcnow()
      dr = DataQuery().time(dt).lonlat_point(-1, -2)
      d = dict(dr)
  
      assert d['time'] == dt.isoformat()
      assert d['latitude'] == -2
      assert d['longitude'] == -1
  
  
  def test_data_query_items():
      """Test the items method of query."""
      dt = datetime.utcnow()
      dr = DataQuery().time(dt).lonlat_point(-1, -2)
      items = list(dr.items())
  
      assert ('time', dt.isoformat()) in items
      assert ('latitude', -2) in items
      assert ('longitude', -1) in items
  
  
  def test_data_query_add():
      """Test adding custom parameters to a query."""
      dr = DataQuery().add_query_parameter(foo='bar')
      assert str(dr) == 'foo=bar'
  
  
  @recorder.use_cassette('gfs-error-no-header')
  def test_http_error_no_header():
      """Test getting an error back without Content-Type."""
      endpoint = HTTPEndPoint('http://thredds.ucar.edu/thredds/ncss/grib/NCEP/GFS/'
                              'Global_0p5deg/GFS_Global_0p5deg_20180223_1200.grib2')
      query = endpoint.query().variables('u-component_of_wind_isobaric')
      query.time(datetime(2018, 2, 23, 22, 28, 49))
      with pytest.raises(HTTPError):
          endpoint.get_query(query)
  
  
  class TestEndPoint(object):
      """Test the HTTPEndPoint."""
  
      def setup(self):
          """Set up tests to point to a common server, api, and end point."""
          self.server = 'http://thredds.ucar.edu/'
          self.api = 'thredds/metadata/grib/NCEP/GFS/Global_0p5deg/TwoD'
          self.endpoint = HTTPEndPoint(self.server + self.api)
  
      def test_basic(self):
          """Test creating a basic query and validating it."""
          q = self.endpoint.query()
          q.all_times()
          assert self.endpoint.validate_query(q), 'Invalid query.'
  
      @recorder.use_cassette('gfs-metadata-map')
      def test_trailing_slash(self):
          """Test setting up and end point with a url with a trailing slash."""
          endpoint = HTTPEndPoint(self.server + self.api + '/')
          q = endpoint.query()
          q.add_query_parameter(metadata='variableMap')
          resp = endpoint.get_query(q)
          assert resp.content
  
      @recorder.use_cassette('gfs-metadata-map')
      def test_query(self):
          """Test getting a query."""
          q = self.endpoint.query()
          q.add_query_parameter(metadata='variableMap')
          resp = self.endpoint.get_query(q)
          assert resp.content
  
      @recorder.use_cassette('gfs-metadata-map-bad')
      def test_get_error(self):
          """Test that getting a bad path raises an error."""
          with pytest.raises(HTTPError):
              self.endpoint.get_path('')
  
      def test_url_path(self):
          """Test forming a url path from the end point."""
          path = self.endpoint.url_path('foobar.html')
          assert path == self.server + self.api + '/foobar.html'
  