Port in the Geo and Users twitter function. Fix issue with handling SpEL expression porperties in the twitter functions.

functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/geo/TwitterGeoFunctionConfiguration.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.twitter.geo;
+
+import java.util.List;
+import java.util.function.Function;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import twitter4j.GeoLocation;
+import twitter4j.GeoQuery;
+import twitter4j.Place;
+import twitter4j.Twitter;
+import twitter4j.TwitterException;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.common.twitter.TwitterConnectionConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+
+/**
+ *
+ * @author Christian Tzolov
+ */
+@Configuration
+@EnableConfigurationProperties(TwitterGeoFunctionProperties.class)
+@Import(TwitterConnectionConfiguration.class)
+public class TwitterGeoFunctionConfiguration {
+
+	private static final Log logger = LogFactory.getLog(TwitterGeoFunctionConfiguration.class);
+
+	@Bean
+	public Function<Message<?>, GeoQuery> messageToGeoQueryFunction(TwitterGeoFunctionProperties geoProperties) {
+		return message -> {
+			String ip = null;
+			if (geoProperties.getSearch().getIp() != null) {
+				ip = geoProperties.getSearch().getIp().getValue(message, String.class);
+			}
+			GeoLocation geoLocation = null;
+			if (geoProperties.getLocation().getLat() != null && geoProperties.getLocation().getLon() != null) {
+				Double lat = geoProperties.getLocation().getLat().getValue(message, Double.class);
+				Double lon = geoProperties.getLocation().getLon().getValue(message, Double.class);
+				geoLocation = new GeoLocation(lat, lon);
+			}
+
+			String query = null;
+			if (geoProperties.getSearch().getQuery() != null) {
+				query = geoProperties.getSearch().getQuery().getValue(message, String.class);
+			}
+			GeoQuery geoQuery = new GeoQuery(query, ip, geoLocation);
+
+			geoQuery.setMaxResults(geoProperties.getMaxResults());
+			geoQuery.setAccuracy(geoProperties.getAccuracy());
+			geoQuery.setGranularity(geoProperties.getGranularity());
+
+			return geoQuery;
+		};
+	}
+
+	@Bean
+	@ConditionalOnProperty(name = "twitter.geo.search.type", havingValue = "search", matchIfMissing = true)
+	public Function<GeoQuery, List<Place>> twitterSearchPlacesFunction(Twitter twitter) {
+		return geoQuery -> {
+			try {
+				return twitter.searchPlaces(geoQuery);
+			}
+			catch (TwitterException e) {
+				logger.error("Places Search failed!", e);
+			}
+			return null;
+		};
+	}
+
+	@Bean
+	@ConditionalOnProperty(name = "twitter.geo.search.type", havingValue = "reverse")
+	public Function<GeoQuery, List<Place>> twitterReverseGeocodeFunction(Twitter twitter) {
+		return geoQuery -> {
+			try {
+				return twitter.reverseGeoCode(geoQuery);
+			}
+			catch (TwitterException e) {
+				logger.error("Reverse Geocode failed!", e);
+			}
+			return null;
+		};
+	}
+
+	@Bean
+	public Function<Message<?>, Message<byte[]>> twitterGeoFunction(
+			Function<Message<?>, GeoQuery> toGeoQuery,
+			Function<GeoQuery, List<Place>> places,
+			Function<Object, Message<byte[]>> managedJson) {
+
+		return toGeoQuery.andThen(places).andThen(managedJson)::apply;
+	}
+}

functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/geo/TwitterGeoFunctionProperties.java

@@ -0,0 +1,197 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.twitter.geo;
+
+import javax.validation.constraints.AssertTrue;
+import javax.validation.constraints.NotNull;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.validation.annotation.Validated;
+
+
+/**
+ * @author Christian Tzolov
+ */
+@ConfigurationProperties("twitter.geo")
+@Validated
+public class TwitterGeoFunctionProperties {
+
+	private static final Expression DEFAULT_EXPRESSION = new SpelExpressionParser().parseExpression("payload");
+
+	public enum GeoType {
+		/** Geo retrieval type. */
+		reverse, search
+	}
+
+	/**
+	 * Geo search API type: reverse or search.
+	 */
+	@NotNull
+	private GeoType type = GeoType.search;
+
+	/**
+	 * Search geo type filter parameters.
+	 */
+	private Search search = new Search();
+
+	/**
+	 *
+	 */
+	private Location location = new Location();
+
+	/**
+	 * Hints for the number of results to return. This does not guarantee that the number of results
+	 * returned will equal max_results, but instead informs how many "nearby" results to return.
+	 */
+	private int maxResults = -1;
+
+	/**
+	 * Sets a hint on the "region" in which to search. If a number, then this is a radius in meters, but it
+	 * can also take a string that is suffixed with ft to specify feet. If this is not passed in, then it is
+	 * assumed to be 0m. If coming from a device, in practice, this value is whatever accuracy the device
+	 * has measuring its location (whether it be coming from a GPS, WiFi triangulation, etc.).
+	 */
+	private String accuracy = null;
+
+	/**
+	 * Minimal granularity of data to return. If this is not passed in, then neighborhood is assumed.
+	 * City can also be passed.
+	 */
+	private String granularity = null;
+
+	public Search getSearch() {
+		return search;
+	}
+
+	public void setSearch(Search search) {
+		this.search = search;
+	}
+
+	public GeoType getType() {
+		return type;
+	}
+
+	public void setType(GeoType type) {
+		this.type = type;
+	}
+
+	public Location getLocation() {
+		return location;
+	}
+
+	public void setLocation(Location location) {
+		this.location = location;
+	}
+
+	public int getMaxResults() {
+		return maxResults;
+	}
+
+	public void setMaxResults(int maxResults) {
+		this.maxResults = maxResults;
+	}
+
+	public String getAccuracy() {
+		return accuracy;
+	}
+
+	public void setAccuracy(String accuracy) {
+		this.accuracy = accuracy;
+	}
+
+	public String getGranularity() {
+		return granularity;
+	}
+
+	public void setGranularity(String granularity) {
+		this.granularity = granularity;
+	}
+
+	@AssertTrue(message = "Either the IP or the Location must be set")
+	public boolean isAtLeastOne() {
+		return this.getSearch().getIp() == null ^ (this.getLocation().getLat() == null && this.getLocation().getLon() == null);
+	}
+
+	@AssertTrue(message = "The IP parameter is applicable only for 'Search' GeoType")
+	public boolean isIpUsedWithSearchGeoType() {
+		if (this.getSearch().getIp() != null) {
+			return this.type == GeoType.search;
+		}
+		return true;
+	}
+
+	public static class Search {
+		/**
+		 * An IP address. Used when attempting to fix geolocation based off of the user's IP address.
+		 * Applicable only for "search" geo type.
+		 */
+		private Expression ip = null;
+
+		/**
+		 * Query expression to filter Places in search results.
+		 */
+		private Expression query = null;
+
+		public Expression getIp() {
+			return ip;
+		}
+
+		public void setIp(Expression ip) {
+			this.ip = ip;
+		}
+
+		public Expression getQuery() {
+			return query;
+		}
+
+		public void setQuery(Expression query) {
+			this.query = query;
+		}
+	}
+
+	public static class Location {
+
+		/**
+		 * User's lat.
+		 */
+		private Expression lat;
+
+		/**
+		 * User's lon.
+		 */
+		private Expression lon;
+
+
+		public Expression getLat() {
+			return lat;
+		}
+
+		public void setLat(Expression lat) {
+			this.lat = lat;
+		}
+
+		public Expression getLon() {
+			return lon;
+		}
+
+		public void setLon(Expression lon) {
+			this.lon = lon;
+		}
+	}
+}

functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/users/TwitterUsersFunctionConfiguration.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.twitter.users;
+
+import java.util.List;
+import java.util.function.Function;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import twitter4j.ResponseList;
+import twitter4j.Twitter;
+import twitter4j.TwitterException;
+import twitter4j.User;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.common.twitter.TwitterConnectionConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+
+/**
+ *
+ * @author Christian Tzolov
+ */
+@Configuration
+@EnableConfigurationProperties(TwitterUsersFunctionProperties.class)
+@Import(TwitterConnectionConfiguration.class)
+public class TwitterUsersFunctionConfiguration {
+
+	private static final Log logger = LogFactory.getLog(TwitterUsersFunctionConfiguration.class);
+
+	@Bean
+	@ConditionalOnProperty(name = "twitter.users.type", havingValue = "search")
+	public Function<Message<?>, List<User>> userSearch(Twitter twitter, TwitterUsersFunctionProperties properties) {
+
+		return message -> {
+			String query = properties.getSearch().getQuery().getValue(message, String.class);
+			try {
+				ResponseList<User> users = twitter.searchUsers(query, properties.getSearch().getPage());
+				return users;
+			}
+			catch (TwitterException e) {
+				logger.error("Twitter API error!", e);
+			}
+			return null;
+		};
+	}
+
+	@Bean
+	@ConditionalOnProperty(name = "twitter.users.type", havingValue = "lookup")
+	public Function<Message<?>, List<User>> userLookup(Twitter twitter, TwitterUsersFunctionProperties properties) {
+
+		return message -> {
+
+			try {
+				TwitterUsersFunctionProperties.Lookup lookup = properties.getLookup();
+				if (lookup.getScreenName() != null) {
+					String[] screenNames = lookup.getScreenName().getValue(message, String[].class);
+					return twitter.lookupUsers(screenNames);
+				}
+				else if (lookup.getUserId() != null) {
+					long[] ids = lookup.getUserId().getValue(message, long[].class);
+					return twitter.lookupUsers(ids);
+				}
+			}
+			catch (TwitterException e) {
+				logger.error("Twitter API error!", e);
+			}
+			return null;
+		};
+	}
+
+	@Bean
+	/**
+	 * queryUsers - depends on the `twitter.users.type` property is either userSearch or userLookup.
+	 * managedJson - converts Users into JSON message payload.
+	 */
+	public Function<Message<?>, Message<byte[]>> twitterUsersFunction(Function<Message<?>, List<User>> queryUsers,
+			Function<Object, Message<byte[]>> managedJson) {
+		return queryUsers.andThen(managedJson);
+	}
+}

functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/users/TwitterUsersFunctionProperties.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.twitter.users;
+
+import javax.validation.constraints.AssertTrue;
+import javax.validation.constraints.NotNull;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * @author Christian Tzolov
+ */
+@ConfigurationProperties("twitter.users")
+@Validated
+public class TwitterUsersFunctionProperties {
+
+	private static final Expression DEFAULT_EXPRESSION = new SpelExpressionParser().parseExpression("payload");
+
+	public enum UserQueryType {
+		/** User retrieval types. */
+		search, lookup
+	}
+
+	/**
+	 * Perform search or lookup type of search.
+	 */
+	@NotNull
+	private UserQueryType type = UserQueryType.search;
+
+	/**
+	 * Returns fully-hydrated user objects for specified by comma-separated values passed to the user_id and/or
+	 * screen_name parameters.
+	 */
+	private final Lookup lookup = new Lookup();
+
+	/**
+	 * relevance-based search interface for querying by topical interest, full name, company name, location,
+	 * or other criteria.
+	 */
+	private final Search search = new Search();
+
+	public UserQueryType getType() {
+		return type;
+	}
+
+	public void setType(UserQueryType type) {
+		this.type = type;
+	}
+
+	public Lookup getLookup() {
+		return lookup;
+	}
+
+	public Search getSearch() {
+		return search;
+	}
+
+	@AssertTrue(message = "Per query type validate the required parameters")
+	public boolean checkParametersPerType() {
+		if (this.getType() == UserQueryType.lookup) {
+			return (this.getLookup().getScreenName() != null) || (this.getLookup().getUserId() != null);
+		}
+		else if (this.getType() == UserQueryType.search) {
+			return (this.getSearch().getQuery() != null);
+		}
+
+		return false;
+	}
+
+	public static class Lookup {
+		/**
+		 * A comma separated list of user IDs, up to 100 are allowed in a single request.
+		 * You are strongly encouraged to use a POST for larger requests.
+		 */
+		private Expression userId;
+
+		/**
+		 * A comma separated list of screen names, up to 100 are allowed in a single request.
+		 * You are strongly encouraged to use a POST for larger (up to 100 screen names) requests.
+		 */
+		private Expression screenName;
+
+		public Expression getUserId() {
+			return userId;
+		}
+
+		public void setUserId(Expression userId) {
+			this.userId = userId;
+		}
+
+		public Expression getScreenName() {
+			return screenName;
+		}
+
+		public void setScreenName(Expression screenName) {
+			this.screenName = screenName;
+		}
+	}
+
+	public static class Search {
+		/**
+		 * The search query to run against people search.
+		 */
+		private Expression query = DEFAULT_EXPRESSION;
+
+		/**
+		 * Specifies the page of results to retrieve.
+		 */
+		private int page = 3;
+
+		public Expression getQuery() {
+			return query;
+		}
+
+		public void setQuery(Expression query) {
+			this.query = query;
+		}
+
+		public int getPage() {
+			return page;
+		}
+
+		public void setPage(int page) {
+			this.page = page;
+		}
+	}
+}

functions/function/twitter-function/src/test/java/org/springframework/cloud/fn/twitter/geo/TwitterGeoFunctionTest.java

@@ -0,0 +1,279 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.twitter.geo;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockserver.client.MockServerClient;
+import org.mockserver.integration.ClientAndServer;
+import twitter4j.conf.ConfigurationBuilder;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.fn.common.twitter.TwitterConnectionProperties;
+import org.springframework.cloud.fn.common.twitter.util.TwitterTestUtils;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.util.MimeTypeUtils;
+import org.springframework.util.SocketUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockserver.matchers.Times.unlimited;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
+import static org.mockserver.verify.VerificationTimes.once;
+
+/**
+ * @author Christian Tzolov
+ */
+@SpringBootTest(
+		webEnvironment = SpringBootTest.WebEnvironment.NONE,
+		properties = {
+				"twitter.connection.consumerKey=consumerKey666",
+				"twitter.connection.consumerSecret=consumerSecret666",
+				"twitter.connection.accessToken=accessToken666",
+				"twitter.connection.accessTokenSecret=accessTokenSecret666"
+		})
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
+public abstract class TwitterGeoFunctionTest {
+
+	private static final String MOCK_SERVER_IP = "127.0.0.1";
+
+	private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort();
+
+	private static ClientAndServer mockServer;
+
+	private static MockServerClient mockClient;
+
+	@Autowired
+	protected Function<Message<?>, Message<byte[]>> twitterUsersFunction;
+
+	public static void recordRequestExpectation(Map<String, List<String>> parameters) {
+
+		mockClient
+				.when(
+						request()
+								.withMethod("GET")
+								.withPath("/geo/search.json")
+								.withQueryStringParameters(parameters),
+						unlimited())
+				.respond(
+						response()
+								.withStatusCode(200)
+								.withHeader("Content-Type", "application/json; charset=utf-8")
+								.withBody(TwitterTestUtils.asString("classpath:/response/search_places_amsterdam.json"))
+								.withDelay(TimeUnit.SECONDS, 1));
+
+	}
+
+	@BeforeAll
+	public static void startMockServer() {
+		mockServer = ClientAndServer.startClientAndServer(MOCK_SERVER_PORT);
+		mockClient = new MockServerClient(MOCK_SERVER_IP, MOCK_SERVER_PORT);
+	}
+
+	@AfterAll
+	public static void stopMockServer() {
+		mockServer.stop();
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.geo.search.ip='127.0.0.1'",
+			"twitter.geo.search.query=payload.toUpperCase()"
+	})
+	public static class TwitterGeoSearchByIPAndQueryTests extends TwitterGeoFunctionTest {
+
+		@Test
+		public void testOne() throws IOException {
+
+			Map<String, List<String>> queryParameters = new HashMap<>();
+			queryParameters.put("ip", Collections.singletonList("127.0.0.1"));
+			queryParameters.put("query", Collections.singletonList("Amsterdam"));
+
+			recordRequestExpectation(queryParameters);
+
+			String inPayload = "Amsterdam";
+
+			Message<?> received = twitterUsersFunction.apply(MessageBuilder.withPayload(inPayload).build());
+
+			mockClient.verify(request()
+							.withMethod("GET")
+							.withPath("/geo/search.json")
+							.withQueryStringParameter("ip", "127.0.0.1")
+							.withQueryStringParameter("query", "AMSTERDAM"),
+					once());
+
+			String outPayload = new String((byte[]) received.getPayload());
+
+			assertThat(outPayload).isNotNull();
+
+			List places = new ObjectMapper().readValue(outPayload, List.class);
+			assertThat(places).hasSize(12);
+		}
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.geo.location.lat='52.378'",
+			"twitter.geo.location.lon='4.9'",
+			"twitter.geo.search.query=payload.toUpperCase()"
+	})
+	public static class TwitterGeoSearchByLocationTests extends TwitterGeoFunctionTest {
+
+		@Test
+		public void testOne() throws IOException {
+
+			Map<String, List<String>> queryParameters = new HashMap<>();
+			queryParameters.put("lat", Collections.singletonList("52.378"));
+			queryParameters.put("long", Collections.singletonList("4.9"));
+			queryParameters.put("query", Collections.singletonList("Amsterdam"));
+
+			recordRequestExpectation(queryParameters);
+
+			String inPayload = "Amsterdam";
+
+			Message<?> received = twitterUsersFunction.apply(MessageBuilder.withPayload(inPayload).build());
+
+			mockClient.verify(request()
+							.withMethod("GET")
+							.withPath("/geo/search.json")
+							.withQueryStringParameter("lat", "52.378")
+							.withQueryStringParameter("long", "4.9")
+							.withQueryStringParameter("query", "Amsterdam"),
+					once());
+
+			String outPayload = new String((byte[]) received.getPayload());
+
+			assertThat(outPayload).isNotNull();
+
+			List places = new ObjectMapper().readValue(outPayload, List.class);
+			assertThat(places).hasSize(12);
+		}
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.geo.type=reverse",
+			"twitter.geo.location.lat='52.378'",
+			"twitter.geo.location.lon='4.9'"
+	})
+	public static class TwitterGeoSearchByLocation2Tests extends TwitterGeoFunctionTest {
+
+		@Test
+		public void testOne() throws IOException {
+
+			Map<String, List<String>> queryParameters = new HashMap<>();
+			queryParameters.put("lat", Arrays.asList("52.378"));
+			queryParameters.put("long", Arrays.asList("4.9"));
+
+			recordRequestExpectation(queryParameters);
+
+			String inPayload = "Amsterdam";
+
+			Message<?> received = twitterUsersFunction.apply(MessageBuilder.withPayload(inPayload).build());
+
+			mockClient.verify(request()
+							.withMethod("GET")
+							.withPath("/geo/search.json")
+							.withQueryStringParameter("lat", "52.378")
+							.withQueryStringParameter("long", "4.9"),
+					once());
+
+			String outPayload = new String((byte[]) received.getPayload());
+
+			assertThat(outPayload).isNotNull();
+
+			List places = new ObjectMapper().readValue(outPayload, List.class);
+			assertThat(places).hasSize(12);
+		}
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.geo.location.lat=#jsonPath(new String(payload),'$.location.lat')",
+			"twitter.geo.location.lon=#jsonPath(new String(payload),'$.location.lon')",
+			"twitter.geo.search.query=#jsonPath(new String(payload),'$.country')"
+	})
+	public static class TwitterGeoSearchJsonPathTests extends TwitterGeoFunctionTest {
+
+		@Test
+		public void testOne() throws IOException {
+
+			Map<String, List<String>> queryParameters = new HashMap<>();
+			queryParameters.put("lat", Collections.singletonList("52.0"));
+			queryParameters.put("long", Collections.singletonList("5.0"));
+			queryParameters.put("query", Collections.singletonList("Netherlands"));
+
+			recordRequestExpectation(queryParameters);
+
+			String inPayload = "{ \"country\" : \"Netherlands\", \"location\" : { \"lat\" : 52.00 , \"lon\" : 5.0 } }";
+
+			Message<?> received = twitterUsersFunction.apply(MessageBuilder
+					.withPayload(inPayload)
+					.setHeader("contentType", MimeTypeUtils.APPLICATION_JSON_VALUE)
+					.build());
+
+			mockClient.verify(request()
+							.withMethod("GET")
+							.withPath("/geo/search.json")
+							.withQueryStringParameter("lat", "52.0")
+							.withQueryStringParameter("long", "5.0")
+							.withQueryStringParameter("query", "Netherlands"),
+					once());
+
+			String outPayload = new String((byte[]) received.getPayload());
+
+			assertThat(outPayload).isNotNull();
+
+			List places = new ObjectMapper().readValue(outPayload, List.class);
+			assertThat(places).hasSize(12);
+		}
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	@Import(TwitterGeoFunctionConfiguration.class)
+	public static class TestTwitterGeoProcessorApplication {
+		@Bean
+		@Primary
+		public twitter4j.conf.Configuration twitterConfiguration2(TwitterConnectionProperties properties,
+				Function<TwitterConnectionProperties, ConfigurationBuilder> toConfigurationBuilder) {
+
+			Function<TwitterConnectionProperties, ConfigurationBuilder> mockedConfiguration =
+					toConfigurationBuilder.andThen(
+							new TwitterTestUtils().mockTwitterUrls(
+									String.format("http://%s:%s", MOCK_SERVER_IP, MOCK_SERVER_PORT)));
+
+			return mockedConfiguration.apply(properties).build();
+		}
+	}
+}

functions/function/twitter-function/src/test/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionTests.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.twitter.trend;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockserver.client.MockServerClient;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.Header;
+import org.mockserver.model.HttpRequest;
+import twitter4j.conf.ConfigurationBuilder;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.fn.common.twitter.TwitterConnectionProperties;
+import org.springframework.cloud.fn.common.twitter.util.TwitterTestUtils;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.util.SocketUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockserver.matchers.Times.exactly;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
+import static org.mockserver.verify.VerificationTimes.once;
+
+/**
+ * @author Christian Tzolov
+ */
+@SpringBootTest(
+		webEnvironment = SpringBootTest.WebEnvironment.NONE,
+		properties = {
+				"twitter.connection.consumerKey=consumerKey666",
+				"twitter.connection.consumerSecret=consumerSecret666",
+				"twitter.connection.accessToken=accessToken666",
+				"twitter.connection.accessTokenSecret=accessTokenSecret666"
+		})
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
+public abstract class TwitterTrendFunctionTests {
+
+	private static final String MOCK_SERVER_IP = "127.0.0.1";
+
+	private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort();
+
+	private static ClientAndServer mockServer;
+
+	private static MockServerClient mockClient;
+	private static HttpRequest trendsRequest;
+
+	@Autowired
+	protected Function<Message<?>, Message<byte[]>> twitterTrendFunction;
+
+	@BeforeAll
+	public static void startServer() {
+		mockServer = ClientAndServer.startClientAndServer(MOCK_SERVER_PORT);
+		mockClient = new MockServerClient(MOCK_SERVER_IP, MOCK_SERVER_PORT);
+
+		trendsRequest = setExpectation(request()
+				.withMethod("GET")
+				.withPath("/trends/place.json")
+				.withQueryStringParameter("id", "2972"));
+	}
+
+	@AfterAll
+	public static void stopServer() {
+		mockServer.stop();
+	}
+
+	public static HttpRequest setExpectation(HttpRequest request) {
+		mockClient
+				.when(request, exactly(1))
+				.respond(response()
+						.withStatusCode(200)
+						.withHeaders(
+								new Header("Content-Type", "application/json; charset=utf-8"),
+								new Header("Cache-Control", "public, max-age=86400"))
+						.withBody(TwitterTestUtils.asString("classpath:/response/trends.json"))
+						.withDelay(TimeUnit.SECONDS, 1)
+				);
+		return request;
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.trend.locationId='2972'",
+			"twitter.connection.rawJson=true"
+	})
+	public static class TwitterTrendPayloadTests extends TwitterTrendFunctionTests {
+
+		@Test
+		public void testOne() {
+			Message<?> received = twitterTrendFunction.apply(MessageBuilder.withPayload("Hello").build());
+			mockClient.verify(trendsRequest, once());
+			assertThat(received).isNotNull();
+		}
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	@Import(TwitterTrendFunctionConfiguration.class)
+	public static class TestTwitterTrendFunctionApplication {
+		@Bean
+		@Primary
+		public twitter4j.conf.Configuration twitterConfiguration2(TwitterConnectionProperties properties,
+				Function<TwitterConnectionProperties, ConfigurationBuilder> toConfigurationBuilder) {
+
+			Function<TwitterConnectionProperties, ConfigurationBuilder> mockedConfiguration =
+					toConfigurationBuilder.andThen(
+							new TwitterTestUtils().mockTwitterUrls(
+									String.format("http://%s:%s", MOCK_SERVER_IP, MOCK_SERVER_PORT)));
+
+			return mockedConfiguration.apply(properties).build();
+		}
+	}
+
+}

functions/function/twitter-function/src/test/java/org/springframework/cloud/fn/twitter/users/TwitterUsersFunctionTests.java

@@ -0,0 +1,204 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.twitter.users;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockserver.client.MockServerClient;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.Header;
+import org.mockserver.model.HttpRequest;
+import twitter4j.conf.ConfigurationBuilder;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.fn.common.twitter.TwitterConnectionProperties;
+import org.springframework.cloud.fn.common.twitter.util.TwitterTestUtils;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.util.SocketUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockserver.matchers.Times.exactly;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
+import static org.mockserver.verify.VerificationTimes.once;
+
+/**
+ * @author Christian Tzolov
+ */
+@SpringBootTest(
+		webEnvironment = SpringBootTest.WebEnvironment.NONE,
+		properties = {
+				"twitter.connection.consumerKey=consumerKey666",
+				"twitter.connection.consumerSecret=consumerSecret666",
+				"twitter.connection.accessToken=accessToken666",
+				"twitter.connection.accessTokenSecret=accessTokenSecret666"
+		})
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
+public abstract class TwitterUsersFunctionTests {
+
+	private static final String MOCK_SERVER_IP = "127.0.0.1";
+
+	private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort();
+
+	private static ClientAndServer mockServer;
+
+	private static MockServerClient mockClient;
+	private static HttpRequest searchUsersRequest;
+	private static HttpRequest lookupUsersRequest;
+	private static HttpRequest lookupUsersRequest2;
+
+	@Autowired
+	protected ObjectMapper mapper;
+
+	@Autowired
+	Function<Message<?>, Message<byte[]>> twitterUsersFunction;
+
+	public static HttpRequest setExpectation(HttpRequest request, String responseUri) {
+		mockClient
+				.when(request, exactly(1))
+				.respond(response()
+						.withStatusCode(200)
+						.withHeaders(
+								new Header("Content-Type", "application/json; charset=utf-8"),
+								new Header("Cache-Control", "public, max-age=86400"))
+						.withBody(TwitterTestUtils.asString(responseUri))
+						.withDelay(TimeUnit.SECONDS, 1)
+				);
+		return request;
+	}
+
+	@BeforeAll
+	public static void startServer() {
+		mockServer = ClientAndServer.startClientAndServer(MOCK_SERVER_PORT);
+		mockClient = new MockServerClient(MOCK_SERVER_IP, MOCK_SERVER_PORT);
+
+		searchUsersRequest = setExpectation(request()
+						.withMethod("GET")
+						.withPath("/users/search.json")
+						.withQueryStringParameter("q", "tzolov")
+						.withQueryStringParameter("page", "3"),
+				"classpath:/response/search_users.json");
+
+		lookupUsersRequest = setExpectation(request()
+						.withMethod("GET")
+						.withPath("/users/lookup.json")
+						.withQueryStringParameter("user_id",
+								"710705860343963648,326896547,267603736,781497571629989888,838754923"),
+				"classpath:/response/lookup_users_id.json");
+
+		lookupUsersRequest2 = setExpectation(request()
+						.withMethod("GET")
+						.withPath("/users/lookup.json")
+						.withQueryStringParameter("screen_name",
+								"TzolovMarto,Rabotnik57,antzolov,peyo_tzolov,ivantzolov"),
+				"classpath:/response/lookup_users_id.json");
+	}
+
+	@AfterAll
+	public static void stopServer() {
+		mockServer.stop();
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.users.type=search",
+			"twitter.users.search.query=payload"
+	})
+	public static class TwitterSearchUsersTests extends TwitterUsersFunctionTests {
+
+		@Test
+		public void testOne() throws IOException {
+			Message<?> received = twitterUsersFunction.apply(MessageBuilder.withPayload("tzolov").build());
+			mockClient.verify(searchUsersRequest, once());
+			assertThat(received).isNotNull();
+			List list = mapper.readValue(new String((byte[]) received.getPayload()), List.class);
+			assertThat(list).hasSize(20);
+		}
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.users.type=lookup",
+			"twitter.users.lookup.userId='710705860343963648,326896547,267603736,781497571629989888,838754923'"
+	})
+	public static class TwitterLookupUserIdLiteralTests extends TwitterUsersFunctionTests {
+
+		@Test
+		public void testOne() throws IOException {
+
+			Message<?> received = twitterUsersFunction.apply(MessageBuilder.withPayload("tzolov").build());
+
+			mockClient.verify(lookupUsersRequest, once());
+			assertThat(received).isNotNull();
+			List list = mapper.readValue(new String((byte[]) received.getPayload()), List.class);
+			assertThat(list).hasSize(5);
+		}
+	}
+
+	@TestPropertySource(properties = {
+			"twitter.users.type=lookup",
+			"twitter.users.lookup.screenName=#jsonPath(payload,'$..[*].code')"
+	})
+	public static class TwitterLookupScreenNamePayloadTests extends TwitterUsersFunctionTests {
+
+		@Test
+		public void testOne() throws IOException {
+
+			Object payload = "[{\"code\":\"TzolovMarto\"},{\"code\":\"Rabotnik57\"},{\"code\":\"antzolov\"},{\"code\":\"peyo_tzolov\"},{\"code\":\"ivantzolov\"}]";
+
+			Message<?> received = twitterUsersFunction.apply(MessageBuilder.withPayload(payload).build());
+			assertThat(received).isNotNull();
+
+			mockClient.verify(lookupUsersRequest2, once());
+			assertThat(received).isNotNull();
+			List list = mapper.readValue(new String((byte[]) received.getPayload()), List.class);
+			assertThat(list).hasSize(5);
+		}
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	@Import(TwitterUsersFunctionConfiguration.class)
+	static class TestTwitterUsersApplication {
+		@Bean
+		@Primary
+		public twitter4j.conf.Configuration twitterConfiguration2(TwitterConnectionProperties properties,
+				Function<TwitterConnectionProperties, ConfigurationBuilder> toConfigurationBuilder) {
+
+			Function<TwitterConnectionProperties, ConfigurationBuilder> mockedConfiguration =
+					toConfigurationBuilder.andThen(
+							new TwitterTestUtils().mockTwitterUrls(
+									String.format("http://%s:%s", MOCK_SERVER_IP, MOCK_SERVER_PORT)));
+
+			return mockedConfiguration.apply(properties).build();
+		}
+	}
+}