tests/auth_test.py

#!/usr/bin/env vpython3
# Copyright 2016 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.

import collections
import contextlib
import datetime
import json
import time

# Mutates sys.path.
import test_env

from utils import tools

tools.force_local_third_party()

# third_party/
import httplib2
from depot_tools import auto_stub

from utils import oauth


class LuciContextAuthTest(auto_stub.TestCase):
  @contextlib.contextmanager
  def lucicontext(self, token_response=None, response_code=200):
    class MockLuciContextServer:
      @staticmethod
      def request(uri=None, method=None, body=None, headers=None):
        self.assertEqual("POST", method)
        self.assertEqual(
          "http://127.0.0.1:0/rpc/LuciLocalAuthService.GetOAuthToken", uri
        )
        self.assertEqual({"Content-Type": "application/json"}, headers)
        data = json.loads(body)
        self.assertEqual("secret", data["secret"])
        self.assertEqual(1, len(data["scopes"]))
        self.assertEqual("acc_a", data["account_id"])
        self.assertEqual(
          "https://www.googleapis.com/auth/userinfo.email", data["scopes"][0]
        )
        response = collections.namedtuple("HttpResponse", ["status"])
        response.status = response_code
        content = json.dumps(token_response)
        return response, content

    self.mock_local_auth(MockLuciContextServer())
    yield

  def setUp(self):
    super(LuciContextAuthTest, self).setUp()
    self.mock_time(0)

  def mock_local_auth(self, server):
    def load_local_auth():
      return oauth.LocalAuthParameters(
        rpc_port=0,
        secret="secret",
        accounts=[oauth.LocalAuthAccount("acc_a")],
        default_account_id="acc_a",
      )

    self.mock(oauth, "_load_local_auth", load_local_auth)

    def http_server():
      return server

    self.mock(httplib2, "Http", http_server)

  def mock_time(self, delta):
    self.mock(time, "time", lambda: delta)

  def _utc_datetime(self, delta):
    return datetime.datetime.utcfromtimestamp(delta)

  def test_get_access_token(self):
    t_expire = 100
    with self.lucicontext({"access_token": "notasecret", "expiry": t_expire}):
      token = oauth._get_luci_context_access_token(oauth._load_local_auth())
    self.assertEqual("notasecret", token.token)
    self.assertEqual(self._utc_datetime(t_expire), token.expires_at)

  def test_get_missing_token(self):
    t_expire = 100
    with self.lucicontext({"expiry": t_expire}):
      token = oauth._get_luci_context_access_token(oauth._load_local_auth())
    self.assertIsNone(token)

  def test_get_missing_expiry(self):
    with self.lucicontext({"access_token": "notasecret"}):
      token = oauth._get_luci_context_access_token(oauth._load_local_auth())
    self.assertIsNone(token)

  def test_get_access_token_with_errors(self):
    with self.lucicontext({"error_code": 5, "error_msg": "fail"}):
      token = oauth._get_luci_context_access_token(oauth._load_local_auth())
    self.assertIsNone(token)

  def test_validation(self):
    t_expire = self._utc_datetime(120)
    good_token = oauth.AccessToken("secret", t_expire)
    self.assertEqual(
      True, oauth._validate_luci_context_access_token(good_token)
    )
    no_token = oauth.AccessToken("", t_expire)
    self.assertEqual(False, oauth._validate_luci_context_access_token(no_token))
    not_a_token = {"token": "secret", "expires_at": t_expire}
    self.assertEqual(
      False, oauth._validate_luci_context_access_token(not_a_token)
    )

    self.mock_time(50)

    self.assertEqual(
      True, oauth._validate_luci_context_access_token(good_token)
    )

    self.mock_time(100)

    self.assertEqual(
      False, oauth._validate_luci_context_access_token(good_token)
    )


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